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:[email protected]: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-8

from 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 = 34
    session.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占位
    # 如果是增刪改，需要 commit
    session.commit()
    # 用完記得關閉，也可以用 with
    session.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()

    # 直接執行SQL
    sql_user()
    sql_user2()

用pycharm運行後的結構

六、小結

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

SQLAlchemy 操作數據庫