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中进行关联查询的几个步骤：

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/Peony-Y/p/5397076.html