查看版本

>>> import sqlalchemy
>>> sqlalchemy.__version__
‘1.0.9‘

創建連接

from sqlclachemy import create_engine
engine = create_engine("sqlite:///:memory:", echo=True)

‘sqlite:///:memory:’ 是 database URL

postgresql

sqlalchemy 默認使用 psycopg2

# 默認情況(即使用psycopg2)
engine = create_engine(‘postgresql://scott:tiger@localhost/mydatabase‘)

# 使用psycopg2
engine = create_engine(‘postgresql+psycopg2://scott:tiger@localhost/mydatabase‘)

# 使用pg8000
engine = create_engine(‘postgresql+pg8000://scott:tiger@localhost/mydatabase‘)
 

MySQL

默認使用mysql-python

# 默認情況（即使用mysql-python）
engine = create_engine(‘mysql://scott:tiger@localhost/foo‘)

# 使用mysql-python
engine = create_engine(‘mysql+mysqldb://scott:tiger@localhost/foo‘)

# 使用MySQL-connector-python
engine = create_engine(‘mysql+mysqlconnector://scott:tiger@localhost/foo‘)

# 使用OurSQL
engine = create_engine(‘mysql+oursql://scott:tiger@localhost/foo‘)
 

Oracle

默認使用cx_oracle

＃ 默認情況（即使用cx_oracle）
engine = create_engine(‘oracle://scott:[email protected]:1521/sidname‘)
# 使用cx_oracle
engine = create_engine(‘oracle+cx_oracle://scott:tiger@tnsname‘)

Microsoft SQL Server

默認使用pyodbc

# 使用pyodbc
engine = create_engine(‘mssql+pyodbc://scott:tiger@mydsn‘)

# 使用pymssql
engine = create_engine(‘mssql+pymssql://scott:tiger@hostname:port/dbname‘)
 

SQLite

因為sqlite是基於文件的數據庫，所以database URL 和前面的不太一樣。

# database URL 形式是 sqlite://<nohostname>/<path>
engine = create_engine(‘sqlite:///foo.db‘)

# 在Unix/Mac
engine = create_engine(‘sqlite:////absolute/path/to/foo.db‘)
# 在Windows
engine = create_engine(‘sqlite:///C:\\path\\to\\foo.db‘)
# 在Windows 中使用原始字符串
engine = create_engine(r‘sqlite:///C:\path\to\foo.db‘)

# 使用內存
engine = create_engine(‘sqlite://‘)
engine = create_engine(‘sqlite:///:memory:‘)

創建表

使用ORM，需要首先描述表，然後把類映射到表上。在SQLAclhemy中這兩步通常在一起，使用叫做Declarative的系統，可以讓我們創建一個包含描述我們需要創建表的類

創建declarative base class

需要先創建基類

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

創建類

比如創建User類，User類需要繼承前面創建的Base。

from sqlalchemy import Column, Integer, String
class User(Base):
    __tablename__ = ‘users‘

    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)
    password = Column(String)

    def __repr__(self):
        return "<User(name=‘%s‘, fullname=‘%s‘, password=‘%s‘)>" % (
                 self.name, self.fullname, self.password)   

需要使用tablename 指名表名。Column 表明字段。

當聲名ORM的類時，Declarative系統會使用python的元類，創建一個包含表信息的Table類（可以通過User.table_ 查看），然後通過Mapper把這個Table與ORM的類聯系起來。所有ORM類的Table類都包含在一個MetaData類中，通過Base.metadata 可以查看MetaData類

對於SQLAlchemy ORM來說Declarative 系統不是必須的，比如：

from sqlalchemy import Table, MetaData, Column, Integer, String, ForeignKey
from sqlalchemy.orm import mapper

metadata = MetaData()

user = Table(‘user‘, metadata,
            Column(‘id‘, Integer, primary_key=True),
            Column(‘name‘, String(50)),
            Column(‘fullname‘, String(50)),
            Column(‘password‘, String(12))
        )

class User(object):
    def __init__(self, name, fullname, password):
        self.name = name
        self.fullname = fullname
        self.password = password

mapper(User, user)

創建User表

Base.metadata.create_all(engine) 

基本的CURD操作

需要創建一個session來對表進行操作

創建一個session

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
Session = sessionmaker()
‘‘‘or
Session = sessionmaker()
Session.configure(bind=engine)
‘‘‘
session = Session()

插入數據

ed_user = User(name=‘ed‘, fullname=‘Ed Jones‘, password=‘edspassword‘)
session.add(ed_user)

一次添加多行

session.add_all([
    User(name=‘wendy‘, fullname=‘Wendy Williams‘, password=‘foobar‘),
    User(name=‘mary‘, fullname=‘Mary Contrary‘, password=‘xxg527‘),
    User(name=‘fred‘, fullname=‘Fred Flinstone‘, password=‘blah‘)])

此時數據是沒有插入到數據庫中的，ed_user.id 是為None的。如果想真正插入數據庫中需要commit。

session.commit()

如果想要操作只是在內存中，不真正commit，ed_user.id 不為None，可以使用flush操作，它只是寫到內存中。

session.flush()

回滾操作

當沒有commit之前，可以回滾

session.rollback()

讀取操作，Querying

執行query操作之前會自動執行flush。
query例子:

for instance in session.query(User).order_by(User.id):
    print(instance.name, instance.fullname)

# 可以只取其中的某些字段
for name, fullname in session.query(User.name, User.fullname):
    print(name, fullname)

for row in session.query(User, User.name).all():
    print(row.User, row.name)

for row in session.query(User.name.label(‘name_label‘)).all():
    print(row.name_label)

可以使用alias對表取別名，可以在表join本身時用。

from sqlalchemy.orm import aliased
user_alias = aliased(User, name=‘user_alias‘)

for row in session.query(user_alias, user_alias.name).all():
    print(row.user_alias)

使用filter_by篩選

for name, in sesson.query(User.name).filter_by(fullname=‘Ed Jones‘):
    print(name)

也可以使用filter，它更加的靈活

for name, in sesson.query(User.name).filter(User.fullname==‘Ed Jones‘):
    print(name)

#相等
query.filter(User.name == ‘ed‘)
#不相等
query.filter(User.name != ‘ed‘)
#like
query.filter(User.name.like(‘%ed%‘))
#in
query.filter(User.name.in_([‘ed‘, ‘wendy‘, ‘jack‘]))
query.filter(User.name.in_(
    session.query(User.name).filter(User.name.like(‘%ed%‘))
))
#not in
query.filter(~User.name.in_([‘ed‘, ‘wendy‘, ‘jack‘]))
#IS NULL
query.filter(User.name == None)
query.filter(User.name.is_(None))
#IS NOT NUKK
query.filter(User.name != None)
query.filter(User.name.isnot(None))
#And
from sqlalchemy import and_
query.filter(and_(User.name == ‘ed‘, User.fullname == ‘Ed Jones‘))
query.filter(User.name == ‘ed‘, User.fullname == ‘Ed Jones‘)
query.filter(User.name == ‘ed‘).filter(User.fullname == ‘Ed Jones‘)
#OR
from sqlalchemy import or_
query.filter(or_(User.name == ‘ed‘, User.name == ‘wendy‘))
#Match
query.filter(User.name.match(‘wendy‘))

返回列表

all() 操作，一次取出所有，返回一個列表。

query = session.query(User).filter(Usre.name.like(‘%ed%‘)).all()

first() 取出第一個,如果沒有就返回None

obj = query.first()

one() 取出一個，如果篩選的結果不只一個，報MultipleResultsFound錯，如果沒有報NoResultFound 錯
one_or_none() 取出一個，如果篩選的結果不只一個，報MultipleResultsFound錯，如果沒有返回None
scalar() 跟one_or_none() 一樣

使用原始的sql

query = session.query(User).filter(‘id<200‘).order_by(text(‘id‘)).all()

使用參數

query = session.query.(User).filter(text(‘id<:value and name=:name‘)).params(value=224, name=‘fred‘).order_by(User.id).one()

使用完全基於string

query = session.query(User).from_statement(
    text(‘SELECT * FROM users where name=:name‘).params(name=‘ed‘).all()
).all()

count

count = session.query(User).filter(User.name.like(‘%ed%‘)).count()

from sqlalchemy import func

query = session.query(func.count(User.name), User.name).group_by(User.name).all()

session.query(func.count(‘*‘)).select_from(User).scalar()

session.query(func.count(User.id)).scalar()

更新

user.name = ‘haha‘
session.commit()

刪除

session.delete(user)

SQLAlchemy數據庫連接和初始化數據庫