Sqlalchemy Docs
Sqlalchemy Docs
Categories:
SQLAlchemy 是python中，通过ORM操作数据库的框架。简单点来说，就是帮助我们从烦冗的sql语句中解脱出来，从而不需要再去写原生的sql语句，只需要用python的语法来操作对象，就能被自动映射为sql语句。
Sqlalchemy 简单使用
安装配置
pip install sqlalchemy
初步使用
1# 导入:
2from sqlalchemy import Column, String, create_engine
3from sqlalchemy.orm import sessionmaker
4from sqlalchemy.ext.declarative import declarative_base
5
6# 创建对象的基类:
7Base = declarative_base()
8
9# 定义User对象:
10class User(Base):
11 # 表的名字:
12 __tablename__ = 'user'
13
14 # 表的结构:
15 id = Column(Integer, primary_key=True, autoincrement=True)
16 name = Column(String(20))
17
18# 初始化数据库连接:
19engine = create_engine('mysql+pymysql://root:root@localhost/tmpdb?charset=utf8')
20# 创建DBSession类型:
21DBSession = sessionmaker(bind=engine)
22# 创建session对象:
23session = DBSession()
24# 创建新User对象:
25new_user = User(id='5', name='Bob')
26# 添加到session:
27session.add(new_user)
28# 提交即保存到数据库:
29session.commit()
30# 关闭session:
31session.close()
初始化数据库连接
1engine = create_engine('dialect+driver://username:password@host:port/database')
2# dialect：数据库类型
3# driver：数据库驱动选择
4# username：数据库用户名
5# password： 用户密码
6# host：服务器地址
7# port：端口
8# database：数据库
1# PostgreSQL
2engine = create_engine('postgresql://username:password@localhost/database')
3engine = create_engine('postgresql+psycopg2://username:password@localhost/database')
4engine = create_engine('postgresql+pg8000://username:password@localhost/database')
5
6# Mysql
7engine = create_engine('mysql://username:password@localhost/database')
8engine = create_engine('mysql+mysqldb://username:password@localhost/database')
9engine = create_engine('mysql+mysqlconnector://username:password@localhost/database')
10engine = create_engine('mysql+oursql://username:password@localhost/database')
11engine = create_engine('mysql+pymysql://username:password@localhost/database')
12
13# Oracle
14engine = create_engine('oracle://username:password@127.0.0.1:1521/database')
15engine = create_engine('oracle+cx_oracle://username:password@tnsname')
16
17# Microsoft SQL Server
18engine = create_engine('mssql+pyodbc://username:password@mydsn')
19engine = create_engine('mssql+pymssql://username:password@hostname:port/database')
20
21# SQLite
22engine = create_engine('sqlite:///database.db')
23engine = create_engine('sqlite:absolute/path/to/database.db')
# create_engine 其它可选参数
# echo ：为 True 时候会把sql语句打印出来
# pool_size： 是连接池的大小，默认为5个，0表示连接数无限制
# pool_recycle： MySQL 默认情况下如果一个连接8小时内容没有任何动作（查询请求）就会自动断开链接，出现 MySQL has gone away的错误。设置了 pool_recycle 后 SQLAlchemy 就会在指定时间内回收连接。如果设置为3600 就表示 1小时后该连接会被自动回收。
# pool_pre_ping ： 这是1.2新增的参数，如果值为True，那么每次从连接池中拿连接的时候，都会向数据库发送一个类似 select 1 的测试查询语句来判断服务器是否正常运行。当该连接出现 disconnect 的情况时，该连接连同pool中的其它连接都会被回收。
MetaData Table
1# 获取元数据
2metadata = MetaData()
3# 定义表
4user = Table('user', metadata,
5 Column('id', Integer, primary_key=True),
6 Column('name', String(20)),
7 )
8
9color = Table('color', metadata,
10 Column('id', Integer, primary_key=True),
11 Column('name', String(20)),
12 )
13
14# 执行sql语句
15engine.execute(
16 "INSERT INTO db_name.color(id, name) VALUES ('1', 'liuyao');"
17)
18result = engine.execute('select * from color')
19print(result.fetchall())
数据模型
1from sqlalchemy.ext.declarative import declarative_base
2from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
3
4Base = declarative_base()
5
6
7class User(Base):
8 __tablename__ = 'user'
9 id = Column(Integer, primary_key=True, autoincrement=True)
10 name = Column(String(80))
11
12 def __repr__(self):
13 return '<User: {}>'.format(self.name)
14
15
16class Address(Base):
17 __tablename__ = 'address'
18 id = Column(Integer, primary_key=True, autoincrement=True)
19 email = Column(String(80))
20 user_id = Column(Integer, ForeignKey('user.id'))
21
22 def __repr__(self):
23 return "<Address %s %d>" % (self.email, self.user_id)
24
25
26# 清除已有表
27Base.metadata.drop_all(engine)
28# 建立表
29Base.metadata.create_all(engine)
30
31
数据查询
1# 前置数据模型代码
2class User(Base):
3 __tablename__ = 'user'
4 id = Column(Integer, primary_key=True, autoincrement=True)
5 name = Column(String(40), unique=True, nullable=False)
6 email = Column(String(80), nullable=True)
7
8 def __repr__(self):
9 return "<User: %s, %s, %s>" % (self.id, self.name, self.email)
10
11 def to_dict(self):
12 return {c.name: getattr(self, c.name, None) for c in self.__table__.columns}
13
14
15class Course(Base):
16 __tablename__ = 'course'
17 id = Column(Integer, primary_key=True, autoincrement=True)
18 name = Column(String(80))
19 # ForeignKey 设置外键关联，第一个参数为字符串，user 为数据表名，id 为字段名
20 # 第二个参数 ondelete 设置删除 User 实例后对关联的 Course 实例的处理规则
21 # 'CASCADE' 表示级联删除，删除用户实例后，对应的课程实例也会被连带删除
22 user_id = Column(Integer, ForeignKey('user.id', ondelete='CASCADE'))
23 # relationship 设置查询接口，以便后期进行数据库查询操作
24 # 第一个参数为位置参数，参数值为外键关联的映射类名，数据类型为字符串
25 # 第二个参数 backref 设置反向查询接口
26 # backref 的第一个参数 'course' 为查询属性，User 实例使用该属性可以获得相关课程实例的列表
27 # backref 的第二个参数 cascade 如此设置即可实现 Python 语句删除用户数据时级联删除课程数据
28 user = relationship('User',
29 backref=backref('course', cascade='all, delete-orphan'))
30
31 def __repr__(self):
32 return '<Course: %s, %s, %s>' % (self.id, self.name, self.user_id)
33
34 def to_dict(self):
35 return {c.name: getattr(self, c.name, None) for c in self.__table__.columns}
36
37
38class Address(Base):
39 __tablename__ = "address"
40 id = Column(Integer, primary_key=True, autoincrement=True, comment='主键')
41 user_id = Column(Integer, nullable=False, comment='用户ID')
42 addr = Column(String(200), nullable=True)
43
44 def __repr__(self):
45 return '<Address %s, %s>' % (self.user_id, self.addr)
46
47 def to_dict(self):
48 return {c.name: getattr(self, c.name, None) for c in self.__table__.columns}
1userquery = session.query(User)
2lst = [
3 # 条件查询
4 userquery.filter(User.id > 3).all(),
5 userquery.filter(User.id > 4).all()[:3],
6 session.query(User.name).filter(User.id > 4).all(),
7 userquery.filter(User.id == 3).first(),
8 userquery.filter(User.id == 5).one(),
9 userquery.filter(User.id.between(3,6)).all(),
10 # 限制返回条数
11 userquery.filter(User.id > 0).all(),
12 userquery.filter(User.id > 0).limit(2).all(),
13 userquery.filter(User.id > 0).offset(2).all(),
14 userquery.filter(User.id > 0).slice(2, 3).all(),
15 # 条件排序
16 userquery.filter(User.id > 0).order_by(User.id).all(),
17 userquery.filter(User.id > 0).order_by(desc(User.id)).all(),
18 userquery.filter(User.id > 0).order_by(User.id.asc()).all(),
19 userquery.filter(User.id > 0).order_by(User.id.desc()).all(),
20 # 不等于
21 userquery.filter(User.id != 3).all(),
22 # 模糊匹配 like, not like
23 userquery.filter(User.email.like('%ao%')).all(),
24 userquery.filter(User.email.notlike('%ao%')).all(),
25 # 匹配
26 userquery.filter(User.email.startswith("a")).all(),
27 userquery.filter(User.email.endswith("g")).all(),
28 userquery.filter(User.email.contains("ao")).all(),
29 # 属于in_ 不属于 notin_
30 userquery.filter(User.id.in_([1, 3, 5])).all(),
31 userquery.filter(User.id.notin_((1, 3, 5))).all(),
32 # 空判断
33 userquery.filter(User.name == None).all(),
34 userquery.filter(User.name.is_(None)).all(),
35 userquery.filter(User.name.isnot(None)).all(),
36 # 多条件
37 userquery.filter(User.id > 3, User.id < 5).all(),
38 userquery.filter(User.id > 3).filter(User.id < 5).all(),
39 # 选择条件
40 userquery.filter(or_(User.id < 2, User.id > 9)).all(),
41 userquery.filter(and_(User.id > 3, User.id < 6)).all(),
42 # 去重
43 userquery.filter(User.id > 3).distinct().all(),
44 # 聚合函数
45 session.query(Course.user_id, func.count(Course.user_id).label('number')).group_by(Course.user_id).all(),
46 session.query(Course.user_id, func.count(Course.user_id).label('number')).group_by(Course.user_id).having(func.count(Course.user_id) > 1).all(),
47 session.query(Course.user_id, func.sum(Course.user_id).label('number')).group_by(Course.user_id).all(),
48 session.query(Course.user_id, func.max(Course.user_id).label('number')).group_by(Course.user_id).all(),
49 session.query(Course.user_id, func.min(Course.user_id).label('number')).group_by(Course.user_id).all(),
50 # 使用extract提取时间中的分钟或者天来分组
51 session.query(extract('minute', User.creatime).label('minute'), func.count('*').label('count')).group_by('minute').all(),
52 session.query(extract('day', User.creatime).label('day'), func.count('*').label('count')).group_by('day').all(),
53 # Join
54 有外键
55 session.query(User.id, User.name.label('uname'), Course.name.label("cname")).join(Course).filter(User.id > 4).all(),
56 无外键, 指定关联关系
57 session.query(User.id, User.name, Address.addr).join(Address, User.id == Address.user_id).filter(User.id > 5).all(),
58 # text 直接写sql
59 userquery.filter(text("id>:id")).params(id=4).all(),
60
61]
62
63# 外连接
64smtp = session.query(User).filter(User.id > 5).subquery()
65lst.append(session.query(smtp.c.name, Address.addr).outerjoin(smtp, Address.user_id == smtp.c.id).all())
66
67
68for res in lst:
69 print(res)
70 # 单对象
71 # data = [r.to_dict() for r in res]
72 # print(data)
73 # 多表查询时
74 # data = [dict(zip(result.keys(), result)) for result in res]
75 # print(data)
76 print("*" * 30)
模型关系
使用代码维护表间外键关联
1class Blog(Base):
2 __tablename__ = 'blog'
3
4 id = Column(BIGINT, primary_key=True, autoincrement=True)
5 title = Column(String(64), server_default='', nullable=False)
6 text = Column(Text, nullable=False)
7 user = Column(BIGINT, nullable=False) # Column(BIGINT, ForeignKey('user.id'), index=True, nullable=False)
8 create = Column(BIGINT, index=True, server_default='0', nullable=False)
9
10 def __repr__(self):
11 return '<Course: %s, %s, %s>' % (self.id, self.title, self.user)
12
13class User(Base):
14 __tablename__ = 'user'
15
16 id = Column(BIGINT, primary_key=True, autoincrement=True)
17 name = Column(String(32), server_default='', nullable=False)
18 username = Column(String(32), index=True, server_default='', nullable=True)
19 password = Column(String(64), server_default='', nullable=False)
20
21 def __repr__(self):
22 return '<Course: %s, %s, %s>' % (self.id, self.name, self.username)
23
24
25# 查询
26print( session.query(User).all() )
27pirnt( session.query(Blog).all() )
28print( session.query(Blog).join(User, user.id==Blog.user).all() )
使用数据库维护表间外键关联
1class Blog(Base):
2 __tablename__ = 'blog'
3
4 id = Column(BIGINT, primary_key=True, autoincrement=True)
5 title = Column(String(64), server_default='', nullable=False)
6 text = Column(Text, nullable=False)
7 user = Column(BIGINT, ForeignKey('user.id'), index=True, nullable=False)
8 create = Column(BIGINT, index=True, server_default='0', nullable=False)
9
10 def __repr__(self):
11 return '<Course: %s, %s, %s>' % (self.id, self.title, self.user)
12
13class User(Base):
14 __tablename__ = 'user'
15
16 id = Column(BIGINT, primary_key=True, autoincrement=True)
17 name = Column(String(32), server_default='', nullable=False)
18 username = Column(String(32), index=True, server_default='', nullable=True)
19 password = Column(String(64), server_default='', nullable=False)
20
21 def __repr__(self):
22 return '<Course: %s, %s, %s>' % (self.id, self.name, self.username)
23
24
25# 查询
26print( session.query(User).all() )
27pirnt( session.query(Blog).all() )
28print( session.query(Blog).join(User).filter(User.id==3).all() )
backref & back_populates
1# backref 只要一边写上就可以
2class User(Base):
3 __tablename__ = 'user'
4 id = Column(Integer, primary_key=True, autoincrement=True)
5 name = Column(String(80))
6
7 addresses = relationship("Address", backref="user")
8
9 def __repr__(self):
10 return '<User: {}>'.format(self.name)
11
12
13class Address(Base):
14 __tablename__ = 'address'
15 id = Column(Integer, primary_key=True, autoincrement=True)
16 email = Column(String(80))
17 user_id = Column(Integer, ForeignKey('user.id'))
18
19 def __repr__(self):
20 return "<Address %s %d>" % (self.email, self.user_id)
21
22# back_populates 要两边都写上
23class User(Base):
24 __tablename__ = 'user'
25 id = Column(Integer, primary_key=True, autoincrement=True)
26 name = Column(String(80))
27
28 addresses = relationship("Address", back_populates="user")
29
30 def __repr__(self):
31 return '<User: {}>'.format(self.name)
32
33
34class Address(Base):
35 __tablename__ = 'address'
36 id = Column(Integer, primary_key=True, autoincrement=True)
37 email = Column(String(80))
38 user_id = Column(Integer, ForeignKey('user.id'))
39
40 user = relationship("User", back_populates="addresses")
41
42 def __repr__(self):
43 return "<Address %s %d>" % (self.email, self.user_id)
44
45# 查询
46
47user = session.query(User).get(3)
48print(user, user.address)
49addr = session.qeury(Address).get(3)
50print(addr, addr.user)
一对多关联
1class Parent(Base): # 一
2 __tablename__ = 'parent'
3 id = Column(Integer, primary_key=True)
4 name = Column(String(64), nullable=False)
5 children = relationship("Child", backref=backref('parents'))
6
7
8class Child(Base): # 多
9 __tablename__ = 'child'
10 id = Column(Integer, primary_key=True)
11 name = Column(String(64), nullable=False)
12 parent_id = Column(Integer, ForeignKey('parent.id'))
13
14# 一对多 一方
15# children = relationship("Child", backref=backref('parents'), cascade="all, delete-orphan")
16# 一对多 多方
17# parents = relationship("Parent", backref=backref('children', cascade="all, delete-orphan"))
18# 以上可以级联删除
一对一关联
1
多对多关联
1