SQLAlchemy 操作數據庫

SQLAlchemy為Python提供了不同數據庫的統一接口，采用ORM的方式操作數據庫，簡潔優雅

一、安裝

直接通過pip安裝即可

pip install sqlalchemy

二、連接數據庫

這里用小巧的sqlite來做測試

from sqlalchemy import create_engine# 創建連接引擎，這里的engine是lazy模式創建，直到第一次被使用才真實創建 # echo=True表示會用logger的方式打印傳到數據庫的SQL engine = create_engine('sqlite:///./test.db', echo=True)

其他數據庫連接方法，格式如下：

數據庫類型+數據庫驅動名稱://用戶名:口令@機器地址:端口號/數據庫名 # 1.1 sqlite內存 engine = create_engine('sqlite:///:memory:', echo=True) # 1.2 sqlite文件 engine = create_engine('sqlite:///./test.db', echo=True)# 2.1 MySQL default engine = create_engine('mysql://user:passwd@localhost/mydatabase') # 2.2 mysql-python engine = create_engine('mysql+mysqldb://user:passwd@localhost/mydatabase') # 2.3 MySQL-connector-python engine = create_engine('mysql+mysqlconnector://user:passwd@localhost/mydatabase') # 2.4 OurSQL engine = create_engine('mysql+oursql://user:passwd@localhost/mydatabase')# 3.1 PostgreSQL default engine = create_engine('postgresql://user:passwd@localhost/mydatabase') # 3.2 psycopg2 engine = create_engine('postgresql+psycopg2://user:passwd@localhost/mydatabase') # 3.3 pg8000 engine = create_engine('postgresql+pg8000://user:passwd@localhost/mydatabase')# 4.1 Oracle default engine = create_engine('oracle://user:passwd@127.0.0.1:1521/mydatabase') # 4.2 cx_oracle engine = create_engine('oracle+cx_oracle://user:passwd@mydatabase')# 5.1 MS SQL pyodbc engine = create_engine('mssql+pyodbc://user:passwd@mydatabase') # 5.2 pymssql engine = create_engine('mssql+pymssql://user:passwd@hostname:port/dbname')

可以看到，SQLAlchemy把數據庫的連接和數據庫的操作分離開來，向上提供統一接口

三、表結構

ORM中，表格對應一個類，所有類都要繼承自Base基類

from sqlalchemy.ext.declarative import declarative_base# 基類 Base = declarative_base()# 定義User對象 class User(Base):"""Users table"""# 表的名字__tablename__ = 'users'__table_args__ = {'sqlite_autoincrement': True}# 表結構id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(32), nullable=False)age = Column(Integer, default=0)password = Column(String(64), unique=True)

四、操作

操作需要通過session來進行，增刪改結束后，需要commit，查詢不用

# 刪除 session = DBSession() duser = session.query(User).filter(User.id==2).delete() session.commit() session.close()#查詢 session = DBSession() quser = session.query(User).filter(User.id==4).one() print('name:',quser.name) session.close()

操作體驗和很多ORM一致

五、完整測試代碼

參考http://www.itfanr.cc/2017/01/06/use-sqlalchemy-by-python/

# coding=utf-8from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String from sqlalchemy.orm import sessionmaker# 創建連接引擎 # echo=True表示會用logger的方式打印傳到數據庫的SQL engine = create_engine('sqlite:///./test.db', echo=True)# 表格對象基類 Base = declarative_base()# 創建會話的類 DBSession = sessionmaker(bind=engine)# 表格類 class User(Base):"""User table"""__tablename__ = 'users' # 表名__table_args__ = {'sqlite_autoincrement': True}# 表結構id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(32), nullable=False) # 有些數據庫允許不指定String的長度age = Column(Integer, default=0)password = Column(String(64), unique=True)class Blog(Base):"""docstring for Blog"""__tablename__ = 'blogs'id = Column(Integer, primary_key=True)title = Column(String(100))desc = Column(String(500))class Tips(Base):"""docstring for tips"""__tablename__ = 'tips'id = Column(Integer, primary_key=True)name = Column(String(32))# 新增一條記錄數據 def new_user():# 創建會話對象session = DBSession()new_user = User(name='Jery',password='123')session.add(new_user) # 只是添加在session中，沒有寫到數據庫session.commit()session.close()# 新增多條數據 def add_more_user():session = DBSession()session.add_all([User(name='guanyu', age=4, password='11111'),User(name='zhangfei', password='2233'),User(name='zhenji', password='44556')])session.commit()session.close()# 新增數據含中文,只要用unicode的字符即可 def add_user_for_zh():session = DBSession()new_user = User(name=u'關羽', password='12322233')session.add(new_user)session.commit()session.close()# 查詢 def query_user():session = DBSession()q_user = session.query(User).filter(User.id == 4).one() # one表示結果有且僅有一個print('name', q_user.name)session.close() # 查詢不用commit，如果不commit，會自動執行rollback# def delete_user():session = DBSession()deleted_num = session.query(User).filter(User.id > 3).delete()print(deleted_num)session.commit()session.close()# c測試 def test_user():session = DBSession()# merge方法，如果存在就修改，不存在就插入（只判斷主鍵，不判斷unique列）t1 = session.query(User).filter(User.name == 'Jery').first()t1.age = 34session.merge(t1)session.commit()# 獲取第2-3項users = session.query(User)[1:3]for u in users:print(u.name)session.close()# 執行sql語句 def sql_user():s = DBSession()# 不能用 `?` 的方式來傳遞參數 要用 `:param` 的形式來指定參數# s.execute('INSERT INTO users (name, age, password) VALUES (?, ?, ?)',('bigpang',2,'1122121'))# 這樣執行報錯# s.execute('INSERT INTO users (name, age, password) VALUES (:aa, :bb, :cc)',({'aa':'bigpang2','bb':22,'cc':'998'}))# s.commit()# 這樣執行成功res = s.execute('select * from users where age=:aaa', {'aaa': 4})# print(res['name']) # 錯誤# print(res.name) # 錯誤# print(type(res)) # 錯誤for r in res:print(r['name'])s.close()# 執行sql語句 def sql_user2():# **傳統 connection方式**# 創建一個connection對象，使用方法與調用python自帶的sqlite使用方式類似# 使用with 來創建 conn，不需要顯示執行關閉連接# with engine.connect() as conn:# res=conn.execute('select * from users')# data=res.fetchone()# print('user is %s' %data[1])# 與python自帶的sqlite不同，這里不需要 cursor 光標，執行sql語句不需要commit。如果是增刪改，則直接生效，也不需要commit.# **傳統 connection 事務**with engine.connect() as conn:trans = conn.begin()try:r1 = conn.execute("select * from users")print(r1.fetchone())r2 = conn.execute("insert into users (name,age,password) values (?,?,?)", ('tang', 5, '133444'))print(r2)trans.commit()except:trans.rollback()raise# **session**session = DBSession()session.execute('select * from users')session.execute('insert into users (name,age,password) values (:name,:age,:password)',{"name": 'dayuzhishui', 'age': 6, 'password': '887'})# 注意參數使用dict，并在sql語句中使用:key占位# 如果是增刪改，需要 commitsession.commit()# 用完記得關閉，也可以用 withsession.close()if __name__ == "__main__":# 刪除全部數據庫Base.metadata.drop_all(engine)# 創建表格，如果已經存在，則不創建Base.metadata.create_all(engine)# 刪除指定的表格Blog.__table__.drop(engine)# 新增數據new_user()# 新增多條數據add_more_user()# 新增數據含中文add_user_for_zh()# 查詢query_user()# 刪除delete_user()# 測試test_user()# 直接執行SQLsql_user()sql_user2()

用pycharm運行后的結構

六、小結

之前遇到過的ORM一般用在web場景，SQLAlchemy作為一個獨立組件而存在，讓我眼前一亮。同時它也十分優雅，既可以使用ORM映射，在必要的場合還可以直接運行SQL，可謂數據處理的利器。

轉載于:https://www.cnblogs.com/fanghao/p/9634173.html

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的SQLAlchemy 操作数据库的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。