Python中應用SQL及SQLAlchemy（一）

以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)

8. 或邏輯（or_）

from sqlalchemy import or_

#查詢title是Engineer或者salary爲3000的記錄
session.query(Account).filter(or_(Account.title=='Engineer', Account.salary==3000))