以SQLit3為例：
import sqlite3 conn = sqlite3.connect('db.sqlite3') #獲取游標對象 cur = conn.cursor() #執行一系列SQL語句 #建立一張表 #cur.execute("create table demo(num int, str vachar(20));") #插入一些記錄 cur.execute("insert into demo values(%d, '%s')" % (1, 'aaa')) cur.execute("insert into demo values(%d, '%s')" % (2, 'bbb')) #更新一條記錄 cur.execute("update demo set str='%s' where num =%d" % ('ddd',3)) #查詢 cur.execute("select * from demo;") rows = cur.fetchall() print("number of records:", len(rows)) for i in rows: print(i) #提交事務 conn.commit() #關閉游標對象 cur.close() #關閉數據庫連接 conn.close()
運行結果：
SQLAlchemy
SQLAlchemy是一款開源軟件，提供了SQL工具包及對象關系映射（ORM）工具，它采用python語言，為高效和高性能的數據庫訪問設計，實現了完整的企業級持久模型，sqlalchemy非常關注數據庫的量級和性能。
使用SQLAlchemy至少需要三部分代碼，這們分別是定義表，定義數據庫連接，進行增、刪、改、查等操作。
創建表的例子：
from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String Base = declarative_base() #定義一個實例，所有表必須繼承該實例 class Account(Base): __tablename__ = 'account' #表名 #字段名 id = Column(Integer, primary_key=True) user_name = Column(String(50), nullable=False) password = Column(String(200), nullable=False) title = Column(String(50)) salary = Column(Integer) def is_active(self): #假設所有用戶都是活躍用戶 return True def get_id(self): #返回帳戶id,該方法返回屬性值提高了表的封裝性 return self.id def is_authenticated(self): #假設已通過驗證 return True def is_anonymous(self): #具有登錄名和密碼的帳戶不是匿名用戶 return False
定義數據庫連接的代碼示例：
from sqlalchemy import create_engine from sqlalchemy.orm import scoped_session, sessionmaker #定義連接數據庫用到的數據庫字符串 db_connect_string = 'mysql://root:123456@localhost:3306/sqlalchemy_db?charset=utf8' #如果數據庫開啟了ssl連接，定義ssl字符串 ssl_args = { 'ssl':{ 'cert': '/home/ssl/client-cert.pem', 'key': '/home/shouse/ssl/client-key.pem', 'ca': '/home/shouse/ssl/ca-cert.pem' } } #如果數據庫開啟了ssl連接，需要傳入ssl engine = create_engine(db_connect_string, connect_args=ssl_args) #定義會話類型 SessionType = scoped_session(sessionmaker(bind=engine, expire_on_commit=False)) def GetSession(): #創建SessionType的實例，用於數據庫操作 return SessionType() from contextlib import contextmanager #定義上下文函數，使能夠自動進行事務處理， #定義上下文件函數的方法就是加上contextmanager裝飾器 #執行邏輯：在函數開始時建立數據庫會話，此時會自動建立一個數據庫事務；當發生異常時回滾（rollback）事務，當 #退出時關閉(close)連接 @contextmanager def session_scope(): session = GetSession() try: yield session session.commit() except: session.rollback() raise finally: session.close()
進行數據庫操作的示例代碼
import orm from sqlalchemy import or_ def InsertAccount(user, password, title, salary): with session_scope() as session: #新增操作 account = orm.Account(user_name=user, password=password, title=title, salary=salary) session.add(account) def CetAccount(id=None, user_name=None): #查詢操作，查詢結果是一個對象集合，同樣可以用all()獲取所有數據 with session_scope() as session: return session.query(orm.Account).filter( or_(orm.Account.id==id, orm.Account.user_name==user_name) ).first() def DeleteAccount(user_name): #刪除操作 with session_scope() as session: account = GetAccount(user_name=user_name) if account: session.delete(account) def UpdateAccount(id, user_name, password, title, salary): #更新操作 with session_scope() as session: account = session.query(orm.Account).filter(orm.Account.id=id).first() if not account: return account.user_name = user_name account.password = password account.title = title account.salary = salary #調用新增操作 InsertAccount('David Li', "123", "System Manager", 3000) InsertAccount('Rebeca Li', '', 'Accountant', 3000) #查詢操作 GetAccount(2) #刪除操作 DeleteAccount('David Li') #更新操作 UpdateAccount(1, "David Li", "none", "System Manager", 2000)
代碼解釋：
- 用import 引入數據庫表Account所在的包orm(orm.py)， 引入多條件查詢時的 或連接 or_
- 每個函數通過with語句啟用上下文函數session_scope(), 通過它獲取到session對象，並自動開啟事務
- 在InsertAccount中，通過新建一個表account實例，並通過session.add將其添加到數據庫中，由於上下文函數退出時會自動提交事務，把以無須顯示地調用session.commit()使新增生
主流數據庫的連接方式
| 數據庫 | 連接字符串 |
| Microsoft SQLServer | ‘mssql+pymssql://username:password@ip:port/dbname’ |
| MySQL | ‘mysql://username:password@ip:port/dbname’ |
| oracle | ‘orcle://username:password@ip:port/dbname’ |
| PostgreSQL | ‘postgresql://username:password@ip:port/dbname’ |
| SQLite | ‘sqlite://file_pathname’ |
查詢條件設置：
在實際編程過程中需要根據各種不同的條件查詢數據庫記錄， SQLAlchemy查詢條件被稱為過濾器。
1. 等值過濾器
session.query(Account).filter(Account.user_name=='Jack') session.query(Account).filter(Account.salary==2000)
2. 不等於過濾器（!=, <, >, <=, >=）
session.query(Account).filter(Account.user_name != 'Jack') session.query(Account).filter(Account.salary != 2000) session.query(Account).filter(Account.salary > 3000)
3. 模糊查詢（like）
模糊查詢只適用於查詢字符串類型，不適用於數值類型
#查詢所有名字中包含字母i的用戶 session.query(Account).filter(Account.user_name.like('%i%')) #查詢所有title中以Manager結尾的用戶 session.query(Account).filter(Account.title.like('%Manager')) #查詢的有名字中以Da開頭的用戶 session.query(Account).filter(Account.user_name.like('Da%'))
4. 包括過濾器(in_)
#查詢id不為1，3，5的記錄 session.query(Account).filter(~Account.id.in_([1,3,5])) #查詢工資不為2000，3000，4000的記錄 session.query(Account).filter(~Account.salary.in_([2000,3000,4000])) #查詢所有title不為Engineer和Accountant的記錄 session.query(Account).filter(~Account.title.in_(['Account','Engineer']))
5. 判斷是否為空（is NULL, is not NULL）
#查詢salary為空值的記錄 session.query(Account).filter(Account.salary.is_(None)) session.query(Account).filter(Account.salary == None) #查詢salary不為空值的記錄 session.query(Account).filter(Account.salary.isnot(None)) session.query(Account).filter(Account.salary != None)
6. 非邏輯 ~
#查詢id不為1，3，5的記錄 session.query(Account).filter(~Account.id.in_([1,3,5]))
7. 與邏輯 （and_）
#直接多個條件查詢 session.query(Account).filter(Account.title='Engineer', Account.salary==3000) #用關鍵字and_進行與邏輯查詢 from sqlalchemy import and_ session.query(Account).filter(and_(Account.title=='Engineer', Account.salary==3000)) #通過多個filter鏈接查詢 session.query(Account).filter(Account.title=='Engineer').filter(Account.salary==3000)
from sqlalchemy import or_ #查詢title是Engineer或者salary為3000的記錄 session.query(Account).filter(or_(Account.title=='Engineer', Account.salary==3000))