SqlAlchemy

SQLAlchemy是Python編程語言下的一款ORM框架，該框架創建在數據庫API之上，使用關係對象映射進行數據庫操做，簡言之即是：將對象轉換成SQL，而後使用數據API執行SQL並獲取執行結果。

Dialect用於和數據API進行交流，根據配置文件的不一樣調用不一樣的數據庫API，從而實現對數據庫的操做，如：

 1 MySQL-Python
 2     mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
 3   
 4 pymysql
 5     mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
 6   
 7 MySQL-Connector
 8     mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
 9   
10 cx_Oracle
11     oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
12   
13 更多詳見：http://docs.sqlalchemy.org/en/latest/dialects/index.html

步驟一：

使用 Engine/ConnectionPooling/Dialect 進行數據庫操做，Engine使用ConnectionPooling鏈接數據庫，而後再經過Dialect執行SQL語句。

 1 from sqlalchemy import create_engine
 2   
 3   
 4 engine = create_engine("mysql+mysqldb://root:123@127.0.0.1:3306/s11", max_overflow=5)
 5   
 6 engine.execute(
 7     "INSERT INTO ts_test (a, b) VALUES ('2', 'v1')"
 8 )
 9   
10 engine.execute(
11      "INSERT INTO ts_test (a, b) VALUES (%s, %s)",
12     ((555, "v1"),(666, "v1"),)
13 )
14 engine.execute(
15     "INSERT INTO ts_test (a, b) VALUES (%(id)s, %(name)s)",
16     id=999, name="v1"
17 )
18   
19 result = engine.execute('select * from ts_test')
20 result.fetchall()

步驟二：

使用 Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 進行數據庫操做。Engine使用Schema Type建立一個特定的結構對象，以後經過SQL Expression Language將該對象轉換成SQL語句，而後經過 ConnectionPooling 鏈接數據庫，再而後經過 Dialect 執行SQL，並獲取結果。

 1 from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey
 2  
 3 metadata = MetaData()
 4  
 5 user = Table('user', metadata,
 6     Column('id', Integer, primary_key=True),
 7     Column('name', String(20)),
 8 )
 9  
10 color = Table('color', metadata,
11     Column('id', Integer, primary_key=True),
12     Column('name', String(20)),
13 )
14 engine = create_engine("mysql+mysqldb://root@localhost:3306/test", max_overflow=5)
15  
16 metadata.create_all(engine)

View Code

增刪改查

 1 from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey
 2  
 3 metadata = MetaData()
 4  
 5 user = Table('user', metadata,
 6     Column('id', Integer, primary_key=True),
 7     Column('name', String(20)),
 8 )
 9  
10 color = Table('color', metadata,
11     Column('id', Integer, primary_key=True),
12     Column('name', String(20)),
13 )
14 engine = create_engine("mysql+mysqldb://root:123@127.0.0.1:3306/s11", max_overflow=5)
15  
16 conn = engine.connect()
17  
18 # 建立SQL語句，INSERT INTO "user" (id, name) VALUES (:id, :name)
19 conn.execute(user.insert(),{'id':7,'name':'seven'})
20 conn.close()
21  
22 # sql = user.insert().values(id=123, name='wu')
23 # conn.execute(sql)
24 # conn.close()
25  
26 # sql = user.delete().where(user.c.id > 1)
27  
28 # sql = user.update().values(fullname=user.c.name)
29 # sql = user.update().where(user.c.name == 'jack').values(name='ed')
30  
31 # sql = select([user, ])
32 # sql = select([user.c.id, ])
33 # sql = select([user.c.name, color.c.name]).where(user.c.id==color.c.id)
34 # sql = select([user.c.name]).order_by(user.c.name)
35 # sql = select([user]).group_by(user.c.name)
36  
37 # result = conn.execute(sql)
38 # print result.fetchall()
39 # conn.close()

View Code

 1 from sqlalchemy import create_engine
 2 from sqlalchemy.ext.declarative import declarative_base
 3 from sqlalchemy import Column, Integer, String
 4 from  sqlalchemy.orm import sessionmaker
 5  
 6 Base = declarative_base() #生成一個SqlORM 基類
 7  
 8  
 9 engine = create_engine("mysql+mysqldb://root@localhost:3306/test",echo=False)
10  
11  
12 class Host(Base):
13     __tablename__ = 'hosts'
14     id = Column(Integer,primary_key=True,autoincrement=True)
15     hostname = Column(String(64),unique=True,nullable=False)
16     ip_addr = Column(String(128),unique=True,nullable=False)
17     port = Column(Integer,default=22)
18  
19 Base.metadata.create_all(engine) #建立全部表結構
20  
21 if __name__ == '__main__':
22     SessionCls = sessionmaker(bind=engine) #建立與數據庫的會話session class ,注意,這裏返回給session的是個class,不是實例
23     session = SessionCls()
24     #h1 = Host(hostname='localhost',ip_addr='127.0.0.1')
25     #h2 = Host(hostname='ubuntu',ip_addr='192.168.2.243',port=20000)
26     #h3 = Host(hostname='ubuntu2',ip_addr='192.168.2.244',port=20000)
27     #session.add(h3)
28     #session.add_all( [h1,h2])
29     #h2.hostname = 'ubuntu_test' #只要沒提交,此時修改也沒問題
30     #session.rollback()
31     #session.commit() #提交
32     res = session.query(Host).filter(Host.hostname.in_(['ubuntu2','localhost'])).all()
33     print(res)

注：SQLAlchemy沒法修改表結構，若是須要能夠使用SQLAlchemy開發者開源的另一個軟件Alembic來完成。

步驟三：

使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 全部組件對數據進行操做。根據類建立對象，對象轉換成SQL，執行SQL。

 1 from sqlalchemy.ext.declarative import declarative_base
 2 from sqlalchemy import Column, Integer, String
 3 from sqlalchemy.orm import sessionmaker
 4 from sqlalchemy import create_engine
 5   
 6 engine = create_engine("mysql+mysqldb://root:123@127.0.0.1:3306/s11", max_overflow=5)
 7   
 8 Base = declarative_base()
 9   
10   
11 class User(Base):
12     __tablename__ = 'users'
13     id = Column(Integer, primary_key=True)
14     name = Column(String(50))
15   
16 # 尋找Base的全部子類，按照子類的結構在數據庫中生成對應的數據表信息
17 # Base.metadata.create_all(engine)
18   
19 Session = sessionmaker(bind=engine)
20 session = Session()
21   
22   
23 # ########## 增 ##########
24 # u = User(id=2, name='sb')
25 # session.add(u)
26 # session.add_all([
27 #     User(id=3, name='sb'),
28 #     User(id=4, name='sb')
29 # ])
30 # session.commit()
31   
32 # ########## 刪除 ##########
33 # session.query(User).filter(User.id > 2).delete()
34 # session.commit()
35   
36 # ########## 修改 ##########
37 # session.query(User).filter(User.id > 2).update({'cluster_id' : 0})
38 # session.commit()
39 # ########## 查 ##########
40 # ret = session.query(User).filter_by(name='sb').first()
41   
42 # ret = session.query(User).filter_by(name='sb').all()
43 # print ret
44   
45 # ret = session.query(User).filter(User.name.in_(['sb','bb'])).all()
46 # print ret
47   
48 # ret = session.query(User.name.label('name_label')).all()
49 # print ret,type(ret)
50   
51 # ret = session.query(User).order_by(User.id).all()
52 # print ret
53   
54 # ret = session.query(User).order_by(User.id)[1:3]
55 # print ret
56 # session.commit()

外鍵關聯

1 from sqlalchemy import Table, Column, Integer, ForeignKey
2 from sqlalchemy.orm import relationship
3 from sqlalchemy.ext.declarative import declarative_base
4 
5 Base = declarative_base()

1 class Parent(Base):
2     __tablename__ = 'parent'
3     id = Column(Integer, primary_key=True)
4     children = relationship("Child")
5  
6 class Child(Base):
7     __tablename__ = 'child'
8     id = Column(Integer, primary_key=True)
9     parent_id = Column(Integer, ForeignKey('parent.id'))

不一樣於上面的一對多，下面增長了語句 relationship

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    children = relationship("Child", back_populates="parent")
 
class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parent.id'))
    parent = relationship("Parent", back_populates="children")

SqlAchemy關聯Join查詢

 

• INNER JOIN: 能夠說是返回兩個表的交集；
• LEFT JOIN: 返回左邊表的全部行，而僅僅返回右邊表和左邊表相匹配的行；
• RIGHT JOIN: 返回右邊表的全部行，而僅僅返回左邊表和右邊表相匹配的行；

具體的區別參見：http://stackoverflow.com/questions/38549/difference-between-inner-and-outer-joins 

group by 查詢：

1 select name,count(host.id) as NumberOfHosts from host right join host_group on host.id= host_group.host_id group by name;

in SQLAchemy

1 from sqlalchemy import func
2 session.query(HostGroup, func.count(HostGroup.name )).group_by(HostGroup.name).all()
3  
4 #another example
5 session.query(func.count(User.name), User.name).group_by(User.name).all() SELECT count(users.name) AS count_1, users.name AS users_name
6 FROM users GROUP BY users.name

SqlAchemy中進行關聯查詢的幾個步驟：