SQLAlchemy

SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。

SQLAlchemy本身不能对数据库进行操作，得依赖第三方的模块，所以还得安装pymysql

Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：

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

步骤一：

使用 Engine/ConnectionPooling/Dialect 进行数据库操作，Engine使用ConnectionPooling连接数据库，然后再通过Dialect执行SQL语句。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
#!/usr/bin/env python
# -*- coding:utf-8 -*-
 
from sqlalchemy import create_engine
 
 
engine = create_engine("mysql+mysqldb://root:123@127.0.0.1:3306/s11", max_overflow=5)
 
engine.execute(
    "INSERT INTO ts_test (a, b) VALUES ('2', 'v1')"
)
 
engine.execute(
     "INSERT INTO ts_test (a, b) VALUES (%s, %s)",
    ((555, "v1"),(666, "v1"),)
)
engine.execute(
    "INSERT INTO ts_test (a, b) VALUES (%(id)s, %(name)s)",
    id=999, name="v1"
)
 
result = engine.execute('select * from ts_test')
result.fetchall()
 事务操作

注：查看数据库连接：show status like 'Threads%';

步骤二：

使用 Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 进行数据库操作。Engine使用Schema Type创建一个特定的结构对象，之后通过SQL Expression Language将该对象转换成SQL语句，然后通过 ConnectionPooling 连接数据库，再然后通过 Dialect 执行SQL，并获取结果。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
#!/usr/bin/env python
# -*- coding:utf-8 -*-
 
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey
 
metadata = MetaData()
 
user = Table('user', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(20)),
)
 
color = Table('color', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(20)),
)
engine = create_engine("mysql+mysqldb://root:123@127.0.0.1:3306/s11", max_overflow=5)
 
metadata.create_all(engine)
# metadata.clear()
# metadata.remove()
 增删改查

更多内容详见：

    http://www.jianshu.com/p/e6bba189fcbd

    http://docs.sqlalchemy.org/en/latest/core/expression_api.html

注：SQLAlchemy无法修改表结构，如果需要可以使用SQLAlchemy开发者开源的另外一个软件Alembic来完成。

步骤三：

使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 所有组件对数据进行操作。根据类创建对象，对象转换成SQL，执行SQL。

1、创建表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
 
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
 
Base = declarative_base()
 
# 创建单表
class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    extra = Column(String(16))
 
    __table_args__ = (
    UniqueConstraint('id', 'name', name='uix_id_name'),
        Index('ix_id_name', 'name', 'extra'),
    )
 
 
# 一对多
class Favor(Base):
    __tablename__ = 'favor'
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default='red', unique=True)
 
 
class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
 
 
# 多对多
class Group(Base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)
    port = Column(Integer, default=22)
 
 
class Server(Base):
    __tablename__ = 'server'
 
    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)
 
 
class ServerToGroup(Base):
    __tablename__ = 'servertogroup'
    nid = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey('server.id'))
    group_id = Column(Integer, ForeignKey('group.id'))
 
 
def init_db():
    Base.metadata.create_all(engine)
 
 
def drop_db():
    Base.metadata.drop_all(engine)

注：设置外检的另一种方式 ForeignKeyConstraint(['other_id'], ['othertable.other_id'])

2、操作表

 表结构 + 数据库连接
  • 增
    复制代码
    obj = Users(name="alex0", extra='sb')
session.add(obj)
session.add_all([
    Users(name="alex1", extra='sb'),
    Users(name="alex2", extra='sb'),
])
session.commit()
    复制代码
  • 删
    session.query(Users).filter(Users.id > 2).delete()
session.commit()
  • 改
    session.query(Users).filter(Users.id > 2).update({"name" : "099"})
session.query(Users).filter(Users.id > 2).update({Users.name: Users.name + "099"}, synchronize_session=False)
session.query(Users).filter(Users.id > 2).update({"num": Users.num + 1}, synchronize_session="evaluate")
session.commit()
  • 查
    ret = session.query(Users).all()
ret = session.query(Users.name, Users.extra).all()
ret = session.query(Users).filter_by(name='alex').all()
ret = session.query(Users).filter_by(name='alex').first()
  • 其他
#　条件
ret = session.query(Users).filter_by(name='alex').all()
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
from sqlalchemy import and_, or_
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
ret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'eric', Users.id > 3),
        Users.extra != ""
    )).all()


# 通配符
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()

# 限制
ret = session.query(Users)[1:2]

# 排序
ret = session.query(Users).order_by(Users.name.desc()).all()
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

# 分组
from sqlalchemy.sql import func

ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()

ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()

# 连表

ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()

ret = session.query(Person).join(Favor).all()#默认是inner join

ret = session.query(Person).join(Favor, isouter=True).all()#left join


# 组合
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all()#去重

q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()#不去重

更多功能参见文档，猛击这里下载PDF

完整代码

创建表

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

Base = declarative_base()

class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    extra = Column(String(16))

class Favor(Base):
    __tablename__ = 'favor'
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default='red', unique=True)


class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))



class Group(Base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True,autoincrement=True)
    name = Column(String(64), unique=True, nullable=False)

class Server(Base):
    __tablename__ = 'server'

    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)


def init_db():
    engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)
#s15day13数据库名 ?charset=utf8支持中文
    Base.metadata.create_all(engine)

init_db()

增

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

Base = declarative_base()

class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    extra = Column(String(16))

engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)

Session = sessionmaker(bind=engine)
session = Session()
#单条数据
# obj = Users(name='alex',extra='lowb')
# session.add(obj)
#多条数据
session.add_all([
    Users(name='alex1',extra='lowb1'),
    Users(name='alex2',extra='lowb2'),
    Users(name='alex3',extra='lowb3'),
])
session.commit()

删

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

Base = declarative_base()

class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    extra = Column(String(16))

engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)

Session = sessionmaker(bind=engine)
session = Session()
session.query(Users).filter(Users.name=='alex').delete()
session.commit()

改

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

Base = declarative_base()

class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    extra = Column(String(16))

engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)

Session = sessionmaker(bind=engine)
session = Session()
#session.query(Users).filter(Users.id==2,Users.name=="alex1").update({Users.name:"ALEX"})
#session.query(Users).filter(Users.id==2,Users.name=="ALEX").update({'name':"ALEX1"})
#session.query(Users).filter(Users.id==2,Users.name=="ALEX1").update({"name":Users.name + 1},synchronize_session="evaluate")
session.query(Users).filter(Users.id > 2).update({Users.name: Users.name + "099"}, synchronize_session=False)

session.commit()

查

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

Base = declarative_base()

class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    extra = Column(String(16))

engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)

Session = sessionmaker(bind=engine)
session = Session()

# 获取全部数据
# v = session.query(Users).all()#v是每个对象生成的列表
# for row in v:
   #通过每个对象对应的方法取具体值
    # print(row.id,row.name,row.extra)

# 获取第一条数据
# v = session.query(Users).first()
# print(v)

# session.query(Users).filter(Users.id==2)
# session.query(Users).filter(Users.id==2,Users.name=="root1")
# session.query(Users).filter(Users.id==2)
#
# v = session.query(Users).filter_by(id=2).all()
# print(v)
# 获取指定映射
v = session.query(Users.id,Users.name)#查询sql语句
print(v)

 连表操作



from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

Base = declarative_base()

class Hobby(Base):
    __tablename__ = 'hobby'
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default='red', unique=True)


class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    hobby_id = Column(Integer, ForeignKey("hobby.nid"))
    hobby = relationship("Hobby", backref='persons')

def init_db():
    engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)
    Base.metadata.create_all(engine)


def drop_db():
    engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)
    Base.metadata.drop_all(engine)

init_db()
engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)

Session = sessionmaker(bind=engine)
session = Session()

# session.add_all([
#     Hobby(nid=1,caption='basketball'),
#     Hobby(nid=2,caption='football'),
#     Hobby(nid=3,caption='banball')
# ])
# session.commit()
#
#
#
# session.add_all([
#     Person(nid=1,name='alex1',hobby_id=1),
#     Person(nid=2,name='alex2',hobby_id=2),
#     Person(nid=3,name='alex3',hobby_id=1),
# ])
# session.commit()
#傻逼版本的连表操作
v = session.query(Person.name,Hobby.caption).join(Hobby,isouter=True).all()
print(v)


 




便捷版本的连表操作


在person表中加了一个字段hobby，仅仅为了方便连表查询


一对多




from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

Base = declarative_base()

class Hobby(Base):
    __tablename__ = 'hobby'
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default='red', unique=True)


class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    hobby_id = Column(Integer, ForeignKey("hobby.nid"))
    hobby = relationship("Hobby", backref='persons')

def init_db():
    engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)
    Base.metadata.create_all(engine)


def drop_db():
    engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)
    Base.metadata.drop_all(engine)

engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)

Session = sessionmaker(bind=engine)
session = Session()

#正向查询，通过person找hobby
v = session.query(Person).all()
for obj in v:
    print(obj.nid,obj.name,obj.hobby.caption)#obj代指Person表的每行，obj.hobby指的就是Hobby这个对象


#反向查询，找出所有爱好是篮球的人名
v = session.query(Person.name,Hobby.caption).join(Hobby,isouter=True).filter(Hobby.caption=='basketball').all()
print(v)
#新方式
v = session.query(Hobby).filter(Hobby.caption=='basketball').all()
for obj in v:
    print(obj.persons,obj.caption)
for row in obj.persons:
    print(row.name)


 




多对多（需要第三表的介入）


傻逼版本




from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

Base = declarative_base()
class Host(Base):
    __tablename__='host'
    nid = Column(Integer,primary_key=True,autoincrement=True)
    hostname = Column(String(32))
    port = Column(String(32))
    ip = Column(String(32))

class HostUser(Base):
    __tablename__ = 'host_user'
    nid = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(32))

class HostToHostUser(Base):
    __tablename__='host_to_hostuser'
    nid = Column(Integer,primary_key=True,autoincrement=True)
    host_id = Column(Integer,ForeignKey('host.nid'))
    host_user_id = Column(Integer,ForeignKey('host_user.nid'))

def init_db():
    engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)
    #t1 库名  ?charset=utf8 支持中文
    Base.metadata.create_all(engine)

engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)

Session = sessionmaker(bind=engine)
session = Session()

# session.add_all([
#         Host(hostname='nginx1',port='22',ip='192.168.1.1'),
#         Host(hostname='nginx2',port='22',ip='192.168.1.2'),
#         Host(hostname='mysql1',port='22',ip='192.168.1.3'),
#         Host(hostname='mysql2',port='22',ip='192.168.1.4'),
#         Host(hostname='git',port='22',ip='192.168.1.5'),
# ])
# session.commit()
# session.add_all([
#         HostUser(username='alex1'),
#         HostUser(username='alex2'),
#         HostUser(username='alex3'),
#
# ])
# session.commit()
# session.add_all([
#         HostToHostUser(host_id=1,host_user_id=1),
#         HostToHostUser(host_id=1,host_user_id=2),
#         HostToHostUser(host_id=2,host_user_id=1),
# ])
# session.commit()

host_obj = session.query(Host).filter(Host.hostname == 'nginx1').first()#根据主机名找到主机id
host_2_host_user = session.query(HostToHostUser.host_user_id).filter(HostToHostUser.host_id == host_obj.nid).all()#根据主机id找到用户id
r = zip(*host_2_host_user)
users = session.query(HostUser.username).filter(HostUser.nid.in_(list(r)[0])).all()#根据用户id，找到用户名
print(users)




 牛逼版本1


通过主机名找到有权限的用户，其实就是利用了relationship,在第三张表中和两个表建立关联，只是查询时需要先通过主机名在host表查到满足的主机对象，而此时主机对象中因为第三张表中的relationship，通过h就能访问hosttohostgroup的对象，再通过第三个表中的第二个relationship正向查询到用户名。




from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

Base = declarative_base()
class Host(Base):
    __tablename__='host'
    nid = Column(Integer,primary_key=True,autoincrement=True)
    hostname = Column(String(32))
    port = Column(String(32))
    ip = Column(String(32))

class HostUser(Base):
    __tablename__ = 'host_user'
    nid = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(32))

class HostToHostUser(Base):
    __tablename__='host_to_hostuser'
    nid = Column(Integer,primary_key=True,autoincrement=True)
    host_id = Column(Integer,ForeignKey('host.nid'))
    host_user_id = Column(Integer,ForeignKey('host_user.nid'))

    host = relationship('Host',backref='h')
    host_user = relationship('HostUser',backref='u')
def init_db():
    engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)
    #t1 库名  ?charset=utf8 支持中文
    Base.metadata.create_all(engine)

engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)

Session = sessionmaker(bind=engine)
session = Session()

host_obj = session.query(Host).filter(Host.hostname=='nginx1').first()
# print(host_obj.nid)
# print(host_obj.hostname)
# print(host_obj.port)
#第三张表对应的对象
# print(host_obj.h)
#循环获取的对象
for item in host_obj.h:
    print(item.host_user.username)




 牛逼版本2




from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index,Table
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

Base = declarative_base()

HostToHostUser = Table('host_to_hostuser',Base.metadata,
                       Column('host_id',ForeignKey('host.nid'),primary_key=True),
                       Column('host_user_id',ForeignKey('host_user.nid'),primary_key=True),)

class Host(Base):
    __tablename__='host'
    nid = Column(Integer,primary_key=True,autoincrement=True)
    hostname = Column(String(32))
    port = Column(String(32))
    ip = Column(String(32))

    host_user = relationship('HostUser',secondary=HostToHostUser,backref='h')

class HostUser(Base):
    __tablename__ = 'host_user'
    nid = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(32))



def init_db():
    engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)
    #t1 库名  ?charset=utf8 支持中文
    Base.metadata.create_all(engine)

engine = create_engine("mysql+pymysql://root:centos@192.168.147.147:3306/s15day13?charset=utf8", max_overflow=5)

Session = sessionmaker(bind=engine)
session = Session()

host_obj = session.query(Host).filter(Host.hostname=='nginx1').first()

print(host_obj.host_user)
for item in host_obj.host_user:
    print(item.username)








        

	      

	    

	  

	  

	    
      
        【推广】
        免费学中医，健康全家人
      
	  

    

    

          原文地址：https://www.cnblogs.com/hongpeng0209/p/6265557.html