SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上,使用关系对象映射进行数据库操作,简言之便是:将类和对象转换成SQL,然后使用数据API执行SQL并获取执行结果。
1 | pip3 install sqlalchemy |
---|
组成部分:
SQLAlchemy本身无法操作数据库,其必须以来pymsql等第三方插件,Dialect用于和数据API进行交流,根据配置文件的不同调用不同的数据库API,从而实现对数据库的操作,如:
SQLAlchemy用一个字符串表示连接信息:
'数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名'
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 |
---|
import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/t1?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程时,最多等待的时间,超时报错,默认30秒
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置),-1代表永远不回收,即一直被重用
)
def task(arg):
conn = engine.raw_connection() #拿到的是一个原生的pymysql连接对象
cursor = conn.cursor()
cursor.execute(
"select * from t1"
)
result = cursor.fetchall()
cursor.close()
conn.close()
for i in range(20):
t = threading.Thread(target=task, args=(i,))
t.start()
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=0, pool_size=5)
def task(arg):
conn = engine.contextual_connect()
with conn:
cur = conn.execute(
"select * from t1"
)
result = cur.fetchall()
print(result)
for i in range(20):
t = threading.Thread(target=task, args=(i,))
t.start()
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine
from sqlalchemy.engine.result import ResultProxy
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=0, pool_size=5)
def task(arg):
cur = engine.execute("select * from t1")
result = cur.fetchall()
cur.close()
print(result)
for i in range(20):
t = threading.Thread(target=task, args=(i,))
t.start()
注意: 查看连接,进程cmd,mysql中>输入 show status like 'Threads%';
import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
Base = declarative_base() # 创建对象的基类:
# 定义User对象:
class Users(Base):
# 表的名字:
__tablename__ = 'users'
# 表的结构:
id = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=False,default='xx') # index指定是否是索引,nullable是否能为空
email = Column(String(32), unique=True) # 指定唯一
ctime = Column(DateTime, default=datetime.datetime.now) #注意,此处设置时datetime.datetime.now若加了括号,则时间永远是程序启动时的时间,后面创建数据时,不会变化
extra = Column(Text, nullable=True)
__table_args__ = (
UniqueConstraint('id', 'name', name='uix_id_name'), # 联合唯一索引
Index('ix_id_name', 'name', 'email'), #给name和email创建普通索引,索引名为ix_id_name
)
def init_db():
"""
根据类创建数据库表
:return:
"""
# 初始化数据库连接:
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
Base.metadata.create_all(engine) #找到所有继承了Base的类,按照其结构建表
def drop_db():
"""
根据类删除数据库表
:return:
"""
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
Base.metadata.drop_all(engine)
if __name__ == '__main__':
drop_db()
init_db()
默认建的表的引擎是MyISAM,如果要设置成InnoDB(支持事务),该怎么设置呢?
__table_args__ = {
'mysql_engine': 'InnoDB', # 指定表的引擎
'mysql_charset': 'utf8' # 指定表的编码格式
}
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime,UniqueConstraint, Index
from sqlalchemy.orm import relationship
Base = declarative_base() # 创建对象的基类:
# ##################### 一对多示例 #########################
class Hobby(Base):
__tablename__ = 'hobby'
id = Column(Integer, primary_key=True)
caption = Column(String(50), default='篮球')
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.id")) #建FK关系
# 与生成表结构无关,仅用于查询方便
hobby = relationship("Hobby", backref='pers') #反向关联的名字
# ##################### 多对多示例 #########################
# 这里多对多需要自己建第三张表,并绑定关系
class Server2Group(Base):
__tablename__ = 'server2group'
id = Column(Integer, primary_key=True, autoincrement=True) #autoincrement 设置自增
server_id = Column(Integer, ForeignKey('server.id'))
group_id = Column(Integer, ForeignKey('group.id'))
class Group(Base):
__tablename__ = 'group'
id = Column(Integer, primary_key=True)
name = Column(String(64), unique=True, nullable=False)
# 与生成表结构无关,仅用于查询方便
servers = relationship('Server', secondary='server2group', backref='groups') #反向关联的名字
class Server(Base):
__tablename__ = 'server'
id = Column(Integer, primary_key=True, autoincrement=True)
hostname = Column(String(64), unique=True, nullable=False)
def init_db():
"""
根据类创建数据库表
:return:
"""
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/userinfo?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
Base.metadata.create_all(engine)
def drop_db():
"""
根据类删除数据库表
:return:
"""
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/userinfo?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
Base.metadata.drop_all(engine)
if __name__ == '__main__':
drop_db()
init_db()
SQLALchemy不同于Django的ORM,当创建多对多关联事,不会自动创建第三张表,需要我们自己定义关系表,进行关联
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine) #创建Session类
# 每次执行数据库操作时,都需要创建一个session
session = Session() # 创建session对象:
# ############# 执行ORM操作 #############
# 创建新User对象
obj1 = Users(name="alex1")
# 添加到session:
session.add(obj1)
# 提交即保存到数据库:
session.commit()
# 关闭session
session.close()
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threading
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
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
# 查询
# cursor = session.execute('select * from users')
# result = cursor.fetchall()
# 添加
cursor = session.execute('insert into users(name) values(:value)',params={"value":'hc'})
# 注意占位符和传参的形式
session.commit()
print(cursor.lastrowid)
session.close()
原生SQL语句
import time
import threading
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
from sqlalchemy.sql import text
from db import Users, Hosts
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
# ################ 添加 ################
obj1 = Users(name="hc")
session.add(obj1) #添加一个对象
session.add_all([
Users(name="hc"),
Users(name="alex"),
Hosts(name="c1.com"),
]) #添加多个对象
session.commit()
# ################ 删除 ################
# filter是where条件,最后调用one()或first()返回唯一行,如果调用all()则返回所有行
session.query(Users).filter(Users.id > 2).delete() #删除Users表中id大于2的数据
session.commit()
# ################ 修改 ################
session.query(Users).filter(Users.id > 0).update({"name" : "099"}) # 将Users表中id>0的数据,把name字段改为099
session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False) #synchronize_session设置为False即执行字符串拼接
session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate") #synchronize_session设置为evaluate即执行四则运算
session.commit()
# ################ 查询 ################
r1 = session.query(Users).all()
r2 = session.query(Users.name.label('xx'), Users.age).all() #label 取别名的,即在查询结果中,显示name的别名'xx'
r3 = session.query(Users).filter(Users.name == "alex").one() # one()返回唯一行,类似于django的get,如果返回数据为多个则报错
r3 = session.query(Users).filter(Users.name == "alex").all() # all()获取所有数据
r4 = session.query(Users).filter_by(name='alex').all() # 注意filter和filter_by后面括号内条件的写法
r5 = session.query(Users).filter_by(name='alex').first() # first()获取返回数据的第一行
r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(Users.id).all()
#order_by后面还可以.desc()降序排列,默认为.asc()升序排列
# text(自定义条件,:的功能类似%s占位),params中进行传参
r7 = session.query(Users).from_statement(text("SELECT * FROM Hosts where name=:name")).params(name='ed').all()
# text中还能从另一个表中查询,前面要用from_statement,而不是filter
session.close()
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threading
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
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Person
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
# 添加
"""
session.add_all([
Hobby(caption='乒乓球'),
Hobby(caption='羽毛球'),
Person(name='张三', hobby_id=3),
Person(name='李四', hobby_id=4),
])
person = Person(name='张九', hobby=Hobby(caption='姑娘'))
session.add(person)
hb = Hobby(caption='人妖')
hb.pers = [Person(name='文飞'), Person(name='博雅')]
session.add(hb) # 会同时创建3条数据(1条hobby的数据,2条person的数据)
session.commit()
"""
# 使用relationship正向查询
"""
v = session.query(Person).first()
print(v.name)
print(v.hobby.caption)
"""
# 使用relationship反向查询
"""
v = session.query(Hobby).first()
print(v.caption)
print(v.pers)
"""
session.close()
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threading
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
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Person, Group, Server, Server2Group
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
# 添加
"""
session.add_all([
Server(hostname='c1.com'),
Server(hostname='c2.com'),
Group(name='A组'),
Group(name='B组'),
])
session.commit()
s2g = Server2Group(server_id=1, group_id=1)
session.add(s2g)
session.commit()
gp = Group(name='C组')
gp.servers = [Server(hostname='c3.com'),Server(hostname='c4.com')]
session.add(gp)
session.commit()
ser = Server(hostname='c6.com')
ser.groups = [Group(name='F组'),Group(name='G组')]
session.add(ser)
session.commit()
"""
# 使用relationship正向查询
"""
v = session.query(Group).first()
print(v.name)
print(v.servers)
"""
# 使用relationship反向查询
"""
v = session.query(Server).first()
print(v.hostname)
print(v.groups)
"""
session.close()
基于relationship操作m2m
and、or、like、limit、排序、分组、连表、组合
# 条件
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()
ret = session.query(Person).join(Favor, isouter=True).all()
# 组合
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()
关联子查询
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threading
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
from sqlalchemy.sql import text, func
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Person, Group, Server, Server2Group
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
# 关联子查询
subqry = session.query(func.count(Server.id).label("sid")).filter(Server.id == Group.id).correlate(Group).as_scalar()
result = session.query(Group.name, subqry)
"""
SELECT `group`.name AS group_name, (SELECT count(server.id) AS sid
FROM server
WHERE server.id = `group`.id) AS anon_1
FROM `group`
"""
# 原生SQL
"""
# 查询
cursor = session.execute('select * from users')
result = cursor.fetchall()
# 添加
cursor = session.execute('insert into users(name) values(:value)',params={"value":'wupeiqi'})
session.commit()
print(cursor.lastrowid)
"""
session.close()
子查询:
session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
"""
select * from users where id in (select id from xxx)
"""
subqry = session.query(func.count(Server.id).label("sid")).filter(Server.id == Group.id).correlate(Group).as_scalar()
#第一步: session.query(func.count(Server.id).label("sid")).filter(Server.id == Group.id)
#这句的sql语句为 select count(id) as sid from server where server.id = group.id 如果直接运行,则会报错
# 第二步:.correlate(Group).as_scalar() ==> 代表此时不执行查询操作,将其当作条件,在group表中查询时,才执行查询
result = session.query(Group.name, subqry)
# sql语句为:select group.name subqry from group
#第三步:将subqry替换为上面的条件,则此句的SQL为:
# select group.name,(select count(id) as sid from server where server.id = group.id) as xx from group
session有两种创建方式
方式一:
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
engine = create_engine(
"mysql+pymysql://root:123@47.93.4.198:3306/s6?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
Session = sessionmaker(bind=engine)
# 方式一:
# 由于无法提供线程共享功能,所有在开发时要注意,在每个线程中自己创建 session。
# from sqlalchemy.orm.session import Session
# 具有操作数据库的:'close', 'commit', 'connection', 'delete', 'execute', 'expire',.....
session = Session() # 创建普通的session
print('原生session',session)
# 操作数据库
session.close()
由于无法提供线程共享功能,所有在开发时要注意,在每个线程中自己创建 session解决办法如下:
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threading
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
from db import Users
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
def task(arg):
session = Session()
obj1 = Users(name="alex1")
session.add(obj1)
session.commit()
for i in range(10):
t = threading.Thread(target=task, args=(i,))
t.start()
方式二(推荐):
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
engine = create_engine(
"mysql+pymysql://root:123@47.93.4.198:3306/s6?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
Session = sessionmaker(bind=engine)
# 方式二:支持线程安全,自动为每个线程创建一个session,单线程时,只创建一个
# - threading.Local
# - 唯一标识
# ScopedSession对象
# self.registry(), 加括号 创建session
# self.registry(), 加括号 创建session
# self.registry(), 加括号 创建session
from greenlet import getcurrent as get_ident #本地线程的唯一标识的函数,加括号则执行函数
session = scoped_session(Session,get_ident)
# session.add
# 操作数据库
session.remove()
支持线程安全,自动为每个线程创建一个session,单线程时,只创建一个
SqlAlchemy本身没有chocie,需要安装这个才能提供choice功能
pip install sqlalchemy-utils
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String
from sqlalchemy_utils import ChoiceType
from sqlalchemy import create_engine
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
type_choices=(
(1,'北京'),
(2,'上海'),
)
id = Column(Integer, primary_key=True) #必须要有主键
name =Column(String(64))
types=Column(ChoiceType(type_choices,Integer())) # 注意:Integer后面要有括号
__table_args__ = {
'mysql_engine': 'InnoDB',
'mysql_charset': 'utf8'
}
def init_db():
"""
根据类创建数据库表
:return:
"""
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/db1?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
Base.metadata.create_all(engine)
def drop_db():
"""
根据类删除数据库表
:return:
"""
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/db1?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间,否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置)
)
Base.metadata.drop_all(engine)
if __name__ == '__main__':
drop_db()
init_db()
#! /usr/bin/env python
# -*- coding: utf-8 -*-
# __author__ = "HuChong"
# Date: 2018/1/12
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from ru import User
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/db1", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
obj1 = User(name="xz",types=1)
obj2 = User(name="zz",types=2)
session.add_all([obj1,obj2])
session.commit()
session.close()
#! /usr/bin/env python
# -*- coding: utf-8 -*-
# __author__ = "HuChong"
# Date: 2018/1/12
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from ru import User
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/db1", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
result_list=session.query(User).all()
print(result_list)
for item in result_list:
print(item.types)
print(item.types.code,item.types.value)
session.close()
#######打印结果如下########
'''
[<ru.User object at 0x0386D770>, <ru.User object at 0x0386D7D0>]
Choice(code=1, value=北京)
1 北京
Choice(code=2, value=上海)
2 上海
'''
用于将Flask和SQLAlchemy联系起来,使用之前需要装下面这个模块
pip install flask-sqlalchemy
如果使用Flask-sqlalchemy组件,则在使用时有一点变化
# 1. 引入Flask-SQLAlchemy
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy() #实例化SQLAlchemy对象
# 2. 注册 Flask-SQLAlchemy
# SQLAlchemy(app)
# 由于这个对象在其他地方想要使用,所有用以下方式注册
db.init_app(app) #读取配置文件,配置文件中写以前在create_engine里面的链接数据
#settings.py中,加上配置
# 3. 导入models中的表
from .models import *
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from app import db
# 4. 写类继承db.Model
class Users(db.Model): #再不是继承Base,而且继承db.Model
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=False)
pwd = Column(String(32))
__table_args__ = {
'mysql_engine': 'InnoDB', # 指定表的引擎
'mysql_charset': 'utf8' # 指定表的编码格式
}
class Group(db.Model):
__tablename__ = 'group'
id = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=False)
__table_args__ = {
'mysql_engine': 'InnoDB',
'mysql_charset': 'utf8'
}
# 5. 创建和删除表
# 以后执行db.create_all()
# 以后执行db.drop_all()
但是这样不好,我们引入 Flask-Migrate
可以通过类似Django里的命令,进行数据迁移,创建表,删除表,更新表
安装 pip install Flask-Migrate
# 5.1 导入
from flask_migrate import Migrate, MigrateCommand
from app import create_app, db
app = create_app()
manager = Manager(app)
# 5.2 创建migrate实例
migrate = Migrate(app, db)
#执行命令:
初次:python manage.py db init
python manage.py db migrate
python manage.py db upgrade
以后执行SQL时:
方式一:
result = db.session.query(models.User.id,models.User.name).all()
db.session.remove()
方式二:
result = models.Users.query.all()
pip freeze # 获取环境中所有安装的模块以及其对应的版本
pip freeze > requirements.txt # 生成对应的文本文件
由于获取的是所有,我们还得自己手动在文本里删除一些不必要的,所有这个方法不好,我们使用下面的方法
pip install pipreqs
首先安装模块,安装完成以后,我们就可以在终端,执行pipreqs命令
# 获取当前所在程序目录中涉及到的所有模块,并自动生成 requirements.txt 且写入内容。
pipreqs ./
建议在Linux系统下使用,windows环境下会报错
以后使用别人的程序,进入程序目录:
安装requirements.txt依赖
pip install -r requirements.txt
会自动安装文件里,所有对应版本模块
https://segmentfault.com/a/1190000003050954