作為一個Pythoner，不會SQLAlchemy都不好意思跟同行打招呼！
SQLAlchemy在Web開發中應用較多，但作為一個數據分析、數據挖掘人員，最好也能掌握這門靈活的資料庫操作技術。它的主要優點有：
- 對數據表的抽象，允許開發人員首先考慮數據模型，同時使得Python程序更加簡潔易讀。
- 對各種資料庫引擎的封裝，使得開發人員在面對不同資料庫時，只需要做簡單修改即可，工作量大大減少。
首先需要利用pip安裝SQLAlchemy：
pip install SQLAlchemy
其次是在程序中通過代碼連接到資料庫：
engine = sqlalchemy.create_engine("mysql+pymysql://username:password@hostname/dbname", encoding="utf8", echo=True)
這裡需要注意，不同的資料庫，不同的資料庫驅動，對應不同的連接字元串。具體可參考：Engine Configuration。
這裡還是以「擼代碼，學知識」的形式學習這門技術。在我的代碼中用到的是MySQL資料庫，以及PyMysql驅動，並盡量多的寫清楚注釋。代碼中主要展示如下功能：
- 創建engine，並通過engine連接資料庫。
- 利用engine連接資料庫後的基本數據操作，包括事務中的批量提交和回滾操作。但這裡沒有用到ORM，不是重點（代碼中利用"""進行了注釋）。
- 利用SQLAlchemy構建數據模型User和Role。這裡兩者存在一對多、多對一的關係。同時利用數據模型構建資料庫表。
- 利用SQLAlchemy中的Session對象進行基礎的數據操作，包括增刪改查等。Session對象可以記錄和跟蹤數據的變化，並提供強大的ORM功能。
- 利用SQLAlchemy中的Session對象進行一些高級數據操作，包括distinct、limit、offset、scalar、count、average、sum、filter、group by、order by等。
- 利用SQLAlchemy中的orm特性試驗外鍵、關聯屬性、join操作等高級操作。這部分建議大家多看多實踐一些，而且重點看一些relationship、backref等關鍵用法，代碼中都有詳細解釋。比如：
- 其他一些小技巧等。
Github上的代碼地址：GitHub - xianhu/LearnPython: 以擼代碼的形式學習Python。文章最後也貼上代碼，方便大家拷貝查看。
另外先做個預告：對於文章一個極為簡潔的Python爬蟲框架，很多人反映沒有實例，我會儘快出一個利用這個框架抓取數據的實例應用。可能是下篇，也可能是下下篇。
=============================================================
作者主頁：笑虎（Python愛好者，關注爬蟲、數據分析、數據挖掘、數據可視化等）
作者專欄主頁：擼代碼，學知識 - 知乎專欄
作者GitHub主頁：擼代碼，學知識 - GitHub
歡迎大家拍磚、提意見。相互交流，共同進步！
==============================================================
# _*_ coding: utf-8 _*_import sqlalchemyimport sqlalchemy.ormimport sqlalchemy.ext.declarative# 利用資料庫字元串構造engine, echo為True將列印所有的sql語句, 其他資料庫的鏈接方式可自行百度# engine = sqlalchemy.create_engine("mysql+pymysql://username:password@hostname/dbname", encoding="utf8", echo=True)engine = sqlalchemy.create_engine("mysql+pymysql://dba_0:mimadba_0@101.200.174.172/data_secret", encoding="utf8", echo=False)"""# 利用engine創建connection,因為使用了with所以不需要close操作,這部分不是重點with engine.connect() as conn: # 最基礎的用法 result = conn.execute("select * from tablename limit 10;") for item in result: print(item) # execute的幾種用法,這裡具體還是得參考pymysql的用法,不需要執行commit操作 conn.execute("insert into tablename(id, url, title) values(1, "url1", "title1");") conn.execute("insert into tablename(id, url, title) values(%s, %s, %s);", 2, "url2", "title2") conn.execute("insert into tablename(id, url, title) values(%s, %s, %s)", (3, "url3", "title3")) conn.execute("insert into tablename(id, url, title) values(%s, %s, %s)", [(31, "url31", "title31"), (32, "url32", "title32")]) # 使用事務可以進行批量提交和回滾 trans = conn.begin() try: conn.execute("insert into tablename(id, url, title) values(%s, %s, %s)", [(4, "url4", "title4"), (5, "url5", "title5")]) trans.commit() except Exception as excep: trans.rollback() raise trans.close()"""# 首先需要生成一個BaseModel類,作為所有模型類的基類BaseModel = sqlalchemy.ext.declarative.declarative_base()# 構建數據模型Userclass User(BaseModel): __tablename__ = "Users" # 表名 __table_args__ = { "mysql_engine": "InnoDB", # 表的引擎 "mysql_charset": "utf8", # 表的編碼格式 } # 表結構,具體更多的數據類型自行百度 id = sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True, autoincrement=True) name = sqlalchemy.Column("name", sqlalchemy.String(50), nullable=False) age = sqlalchemy.Column("age", sqlalchemy.Integer, default=0) # 添加角色id外鍵(關聯到Role.id屬性) role_id = sqlalchemy.Column("role_id", sqlalchemy.Integer, sqlalchemy.ForeignKey("Roles.id")) # 添加關係屬性(關聯到role_id外鍵上) role = sqlalchemy.orm.relationship("Role", foreign_keys="User.role_id") # 添加關係屬性(關聯到role_id外鍵上),如果使用了這種方式,Role模型中的users可以省略 # role = sqlalchemy.orm.relationship("Role", foreign_keys="User.role_id", backref=sqlalchemy.orm.backref("users"))# 構建數據模型Roleclass Role(BaseModel): __tablename__ = "Roles" # 表名 __table_args__ = { "mysql_engine": "InnoDB", # 表的引擎 "mysql_charset": "utf8", # 表的編碼格式 } # 表結構,具體更多的數據類型自行百度 id = sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True) name = sqlalchemy.Column("name", sqlalchemy.String(50), unique=True) # 添加關係屬性(關聯到User.role_id屬性上) users = sqlalchemy.orm.relationship("User", foreign_keys="User.role_id")# 利用Session對象連接資料庫DBSessinon = sqlalchemy.orm.sessionmaker(bind=engine) # 創建會話類session = DBSessinon() # 創建會話對象# 刪除所有表BaseModel.metadata.drop_all(engine)# 創建所有表,如果表已經存在,則不會創建BaseModel.metadata.create_all(engine)try: # 清空數據,不需要commit操作 session.query(User).filter(User.id != -1).delete() session.query(Role).filter(Role.id != -1).delete() # 刪除數據的另外一種形式:session.delete() # 插入數據,這裡的一個實例只插入一次,第二次插入不生效 session.add(Role(id=1, name="student")) session.add(Role(id=2, name="teacher")) session.commit() session.add(User(name="James", age=20, role_id=1)) session.add(User(name="Wade", age=40, role_id=2)) session.commit() user = User(name="Kobe", age=24, role_id=1) session.add(user) session.commit() # 修改數據 user.name = "Allen" session.merge(user) # 使用merge方法,如果存在則修改,如果不存在則插入 session.query(User).filter(User.id == user.id).update({User.name: "Allen"}) # 使用update方法 session.query(User).filter(User.id == user.id).update({User.age: User.age + 1}) # 使用update方法,自增操作 # 查詢數據 roles = session.query(Role) # 返回全部結果 for role in roles: print("Role:", role.id, role.name) users = session.query(User) # 返回全部結果 for user in users: print("User:", user.id, user.name, user.age, user.role_id) # 其他獲取數據的方式 print("get(id):", session.query(User).get(1)) # 返回結果集中id為1的項 print("get[1:3]:", session.query(User)[1:3]) # 返回結果集中的第2-3項 # 其他高級查詢,這裡以Users表為例 users = session.query(User).filter(User.id > 6) # 條件查詢 users = session.query(User).filter(User.id > 6).all() # 條件查詢,返回查詢的全部數據 user = session.query(User).filter(User.id > 6).first() # 條件查詢,返回查詢數據的第一項 users = session.query(User).filter(User.id > 6).limit(10) # 條件查詢,返回最多10條數據 users = session.query(User).filter(User.id > 6).offset(2) # 條件查詢,從第3條數據開始返回 users = session.query(User).filter(User.id > 6, User.name == "Kobe") # 條件查詢,and操作 users = session.query(User).filter(User.id > 6).filter(User.name == "Kobe") # 條件查詢,and操作 users = session.query(User).filter(sqlalchemy.or_(User.id > 6, User.name == "Kobe")) # 條件查詢,or操作 users = session.query(User).filter(User.id.in_((1, 2))) # 條件查詢,in操作 users = session.query(User).filter(sqlalchemy.not_(User.name)) # 條件查詢,not操作 user_count = session.query(User.id).count() # 統計全部user的數量 user_count = session.query(sqlalchemy.func.count(User.id)).scalar() # scalar操作返回第一行數據的第一個欄位 session.query(sqlalchemy.func.count("*")).select_from(User).scalar() # scalar操作返回第一行數據的第一個欄位 session.query(sqlalchemy.func.count(1)).select_from(User).scalar() # scalar操作返回第一行數據的第一個欄位 session.query(sqlalchemy.func.count(User.id)).filter(User.id > 0).scalar() # filter() 中包含 User，因此不需要指定表 session.query(sqlalchemy.func.sum(User.age)).scalar() # 求和運算,運用scalar函數 session.query(sqlalchemy.func.avg(User.age)).scalar() # 求均值運算,運用scalar函數 session.query(sqlalchemy.func.md5(User.name)).filter(User.id == 1).scalar() # 運用md5函數 users = session.query(sqlalchemy.distinct(User.name)) # 去重查詢,根據name進行去重 users = session.query(User).order_by(User.name) # 排序查詢,正序查詢 users = session.query(User).order_by(User.name.desc()) # 排序查詢,倒序查詢 users = session.query(User).order_by(sqlalchemy.desc(User.name)) # 排序查詢,倒序查詢的另外一種形式 users = session.query(User.id, User.name) # 只查詢部分屬性 users = session.query(User.name.label("user_name")) # 結果集的列取別名 for user in users: print("label test:", user.user_name) # 這裡使用別名 users = session.query(sqlalchemy.func.count(User.name).label("count"), User.age).group_by(User.age) # 分組查詢 for user in users: print("age:{0}, count:{1}".format(user.age, user.count)) # 多表查詢 result = session.query(User, Role).filter(User.role_id == Role.id) for user, role in result: print("user %s"s role is %s" % (user.name, role.name)) users = session.query(User).join(Role, User.role_id == Role.id) for user in users: print("user join, name:", user.name) # 關聯屬性的用法 roles = session.query(Role) for role in roles: print("role:%s users:" % role.name) for user in role.users: print(" %s" % user.name) users = session.query(User) for user in users: print("user %s"s role is %s" % (user.name, user.role.name))except Exception as excep: session.rollback() raisesession.close()
推薦閱讀：
※大數據精準營銷｜如何與用戶談一場不分手的戀愛？
※Python數據分析之基情的擇天記
※數據 | 沒錢沒資源，怎麼做調研（一）
※航班航行距離與延誤時間有什麼關係？
※股票分析與資產組合（python）