sqlalchemy基本增删查改

sqlalchemy介绍和快速使用

1.sqlalchemy:orm框架
    -django orm：只能给django用，不能独立用
    -sqlalchemy：独立使用，集成到web项目中
    -peewee：小
    -tortoise-orm ：异步orm框架

2.安装
pip3 install sqlalchemy

3.组成部分
Engine，框架的引擎
Connection Pooling ，数据库连接池
Dialect，选择连接数据库的DB API种类：mysql，sqllite。。。
Schema/Types，架构和类型
SQL Exprression Language，SQL表达式语言

4.能够操作的关系型数据库
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

原生操作的快速使用

-使用步骤
1 第一步：导入包
from threading import Thread
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine

2 第二步：实例化得到一个engine
engine = create_engine(
    "mysql+pymysql://root:[email protected]:3306/luffy?charset=utf8",
    max_overflow=0,  # 超过连接池大小外最多创建的连接
    pool_size=5,  # 连接池大小
    pool_timeout=30,  # 池中没有，线程最多等待的时间，否则报错
    pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置），-1表示不会回收
)

3 第三步：通过engine拿到一个链接
# 拿到一个conn对象，从连接池中取出一个链接
def task():
    conn = engine.raw_connection()
    cursor = conn.cursor()
    cursor.execute("select * from luffy_banner")
    print(cursor.fetchall())

4 第三步：多线程测试
for i in range(20):
    t=Thread(target=task)
    t.start()

创建操作数据表

1.通过类 创建和删除表
-第一步：导入一些依赖
-第二步：创建成一个Base：Base = declarative_base()
-第三步：写类：都集成Base
class User(Base):
-第四步：写字段，字段都是Column类的对象，通过参数控制字段类型，是否可以为空，是否索引。。。
    id = Column(Integer, primary_key=True)  # id 主键
    name = Column(String(32), index=True, nullable=False)
-第五步：定义表名，联合唯一，联合索引
    __tablename__ = 'users'  # 数据库表名称
    # 定义联合索引，联合唯一
    __table_args__ = (
        UniqueConstraint('id', 'name', name='uix_id_name'),  # 联合唯一
        Index('ix_id_name', 'name', 'email'),  # 联合索引
    )
-第六步：把被Base管理的所有表，同步到数据库中[不能创建数据库，不能删除修改字段]
    engine = create_engine(
        "mysql+pymysql://root:[email protected]:3306/aaa?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )
    # 创建出所有被Base管理的表
    Base.metadata.create_all(engine)
-第七步：删除被Base管理的所有表
    # 删除所有被Base管理的表
    Base.metadata.drop_all(engine)

2.实例
# 写一个个类，继承某个父类，写字段
# 第一步：导入一些依赖
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
import datetime
from sqlalchemy import create_engine

# 第二步：创建一个父类
Base = declarative_base()


# 第三步：写类，继承父类
class User(Base):
    # 第四步:写字段，所有字段都是Column的对象，在里面通过参数控制类型
    id = Column(Integer, primary_key=True)  # id 主键
    name = Column(String(32), index=True, nullable=False)  # varchar32  name列，索引，不可为空
    email = Column(String(32), unique=True)  # 唯一
    # datetime.datetime.now不能加括号，加了括号，以后永远是当前时间
    ctime = Column(DateTime, default=datetime.datetime.now)
    extra = Column(Text, nullable=True)

    # 定义表名字
    __tablename__ = 'users'  # 数据库表名称
    # 定义联合索引，联合唯一
    __table_args__ = (
        UniqueConstraint('id', 'name', name='uix_id_name'),  # 联合唯一
        Index('ix_id_name', 'name', 'email'),  # 联合索引
    )


class Book(Base):
    __tablename__ = 'books'  # 数据库表名称
    id = Column(Integer, primary_key=True)  # id 主键
    name = Column(String(32), index=True, nullable=False)  # varchar32  name列，索引，不可为空
    price = Column(Integer)

    def __repr__(self):
        return str(self.price)


class Publish(Base):
    __tablename__ = 'publish'  # 数据库表名称
    id = Column(Integer, primary_key=True)  # id 主键
    name = Column(String(32), nullable=True)

# 第5步：sqlalchemy没有迁移一说，只能创建出被Base管理的所有表，和删除被Base管理的所有表
# sqlalchemy不能创建数据库，不能修改，删除字段，只能创建表，和删除表
def init_db():
    engine = create_engine(
        "mysql+pymysql://root:[email protected]:3306/aaa?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )
    # 创建出所有被Base管理的表
    Base.metadata.create_all(engine)


def drop_db():
    engine = create_engine(
        "mysql+pymysql://root:[email protected]:3306/aaa?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )
    # 删除所有被Base管理的表
    Base.metadata.drop_all(engine)


if __name__ == '__main__':
    init_db()
    # drop_db()

sqlalchemy快速插入数据

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import User, Book

# 第一步：创建engine
engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa", max_overflow=0, pool_size=5)

# 第二步：通过engine，获得session对象：跟之前学的cookie，session不是一个东西
Session = sessionmaker(bind=engine)
# 每次执行数据库操作时，都需要创建一个Connection
session = Session()

# 第三步，通过session操作插入数据
# book=Book(name='西游记',price=33)
user = User(name='barry', email='[email protected]', extra='很帅')
session.add(user)
session.commit()
session.close()

scoped_session线程安全

-session是链接对象，如果集成到flask中，我们是吧session定义成全局，还是每个视图函数一个session呢？正常来讲要每个视图函数定义一个session，有些麻烦
-sqlalchemy 帮咱提供了一个只要定义一次的session，能够做到在不同线程中，使用的是自己的session，底层基于local

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import User, Book
from sqlalchemy.orm import scoped_session
from threading import Thread

# 第一步：创建engine
engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa", max_overflow=0, pool_size=5)

# 第二步：通过engine，获得session对象：跟之前学的cookie，session不是一个东西
Session = sessionmaker(bind=engine)

# 原来
# session=Session() #不是线程安全
# 以后咱们使用这个它做到了线程安全
session = scoped_session(Session)


def task(i):
    user = User(name='彭于晏%s' % i, email='%[email protected]' % i, extra='很丑')
    session.add(user)
    session.commit()
    session.close()


for i in range(20):
    t = Thread(target=task, args=[i, ])
    t.start()

类装饰器

-类装饰器之：加在类上的装饰器
def auth(func):
    def inner(*args, **kwargs):
        print("我要开始了")
        res = func(*args, **kwargs)
        res.name = 'lqz'
        return res
    return inner

@auth  # Person=auth(Person)  # 加在类上的装饰器
class Person():
    pass

p = Person()
print(p.name)


-类装饰器之：装饰器是类
class Auth():
    def __init__(self, func):
        self.func = func

    def __call__(self, a, b):
        print('我要开始加了')
        res = self.func(a, b)
        return res

@Auth  # add=Auth(add)
def add(a, b):  # add 是Auth的对象
    return a + b

res=add(4,5)  # add加括号，会触发Auth的__call__
print(res)

一对多

表模型

from sqlalchemy.orm import relationship

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指的是tablename而不是类名，uselist=False
    hobby_id = Column(Integer, ForeignKey("hobby.id"))
    
    # 跟数据库无关，不会新增字段，只用于快速链表操作
    # 类名，backref用于反向查询
    hobby=relationship('Hobby',backref='pers')

新增和基于对象的查询

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Person, Hobby
from sqlalchemy.orm import scoped_session

engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)

1 一对多关系新增
hobby = Hobby(caption='足球')
# 把hobby存入
session.add(hobby)
print(hobby.id)

-新增方案
# 方案一
person = Person(name='lqz', hobby_id=1)
session.add(person)

# 方案二：
person = Person(name='张三', hobby=Hobby(caption='乒乓球'))
session.add(person)

# 方案三：
hobby = session.query(Hobby).filter_by(id=2).first()
print(hobby)
person = Person(name='李四', hobby=hobby)
person = Person(name='王五', hobby_id=hobby.id)
session.add(person)

2 基于对象的跨表查询
# 基于对象的跨表查询的正向查询
person = session.query(Person).filter_by(id=1).first()
print(person.hobby_id)
print(person.hobby.caption)

-基于对象的跨表查的反向
hobby = session.query(Hobby).filter_by(id=2).first()
# 喜欢这个爱好的所有人
print(hobby.pers)  # 列表
for p in hobby.pers:
    print(p.name)

session.commit()
session.close()

多对多

表模型

class Boy2Girl(Base):
    __tablename__ = 'boy2girl'
    id = Column(Integer, primary_key=True, autoincrement=True)
    girl_id = Column(Integer, ForeignKey('girl.id'))
    boy_id = Column(Integer, ForeignKey('boy.id'))


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

    def __str__(self):
        return self.name

    def __repr__(self):
        return self.name


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

    # 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以
    # 方便快速查询，写了这个字段，相当于django 的manytomany，快速使用基于对象的跨表查询
    girls = relationship('Girl', secondary='boy2girl', backref='boys')

    def __str__(self):
        return self.name

    def __repr__(self):
        return self.name

增加和基于对象的跨表查询

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Boy, Girl, Boy2Girl
from sqlalchemy.orm import scoped_session

engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)

1 多对多新增
手动操作第三张表的方式
boy = Boy(name='彭于晏')
girl = Girl(name='刘亦菲')
session.add(boy)
session.add(girl)
session.add_all([boy, girl])
建立关系：手动操作第三张表
b = Boy2Girl(girl_id=1, boy_id=1)
session.add(b)

2 通过关联关系
session.add(Boy(name='李清照', girls=[Girl(name='小红'), Girl(name='小黄')]))
# 拆开
girl1 = Girl(name='小红1')
girl2 = Girl(name='小黄1')
boy = Boy(name='李清照1', girls=[girl2, girl1])
session.add(boy)

3 基于对象的跨表查
正向
boy = session.query(Boy).filter_by(id=2).first()
print(boy)
# 跟这个男孩越过的所有女生
print(boy.girls)

4 通过girl查boy
# 反向
girl = session.query(Girl).filter_by(id=2).first()
print(girl)
print(girl.boys)

session.commit()
session.close()

基本增删查改

基本增删查改和高级查询

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Boy, Girl, Boy2Girl, Hobby, Book, User, Person
from sqlalchemy.orm import scoped_session
from sqlalchemy.sql import text

engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)

1 基本增
# add    add_all
# Hobby表增加记录
hobby = Hobby(caption='羽毛球')
book = Book(name='三国演义', price=22)
# session.add(hobby)
session.add_all([hobby, book])

2 删除--->查询再删---》一般写项目都是软删除
hobby = session.query(Hobby).filter_by(caption='足球').delete()
print(hobby)

3 修改，更新
res = session.query(User).filter(User.id > 0).update({"name": "barry"})
# 类似于django的F查询
# synchronize_session=False当字符串相加
res = session.query(User).filter(User.id > 0).update({User.name: User.name + "099"}, synchronize_session=False)
# synchronize_session="evaluate数字相加
res = session.query(Book).filter(Book.price > 20).update({"price": Book.price + 1}, synchronize_session="evaluate")
print(res)

4 查询： filer:写条件     filter_by：等于的值
4.1 查询所有  是list对象
res = session.query(User).all()
print(type(res))
print(len(res))
4.1.1 只查询某几个字段
# select name as xx,email from user;
res = session.query(User.name.label('xx'), User.email)
print(res)  # 打出原生sql
print(res.all())
for item in res.all():
    print(item[0])
4.1.2 filter传的是表达式，filter_by传的是参数
res = session.query(User).filter(User.name == "barry").all()
res = session.query(User).filter(User.name != "barry").all()
res = session.query(User).filter(User.name != "barry", User.email == '[email protected]').all()
res = session.query(User).filter_by(name='barry099').all()
res = session.query(User).filter_by(name='barry099', email='[email protected]').all()
print(len(res))

4.2 取一个 all了后是list，list 没有first方法
res = session.query(User).first()
print(res)

4.3 查询所有，使用占位符  :value     :name
res = session.query(User).filter(text("id<:value and name=:name")).params(value=10, name='barry099').order_by(
    User.id).all()
print(len(res))

4.4 自定义查询
res = session.query(User).from_statement(text("SELECT * FROM users where email=:email")).params(email='[email protected]').all()
print(res)


4.5 高级查询
-条件
表达式，and条件连接
res = session.query(User).filter(User.id > 1, User.name == 'barry099').all() # and条件
between
res = session.query(User).filter(User.id.between(1, 3), User.name == 'barry099').all()
# in
res = session.query(User).filter(User.id.in_([1, 3, 4])).all()
res = session.query(User).filter(User.email.in_(['[email protected]', '[email protected]'])).all()

# ~非，除。。外
res = session.query(User).filter(~User.id.in_([1, 3, 4])).all()

-二次筛选
res = session.query(User).filter(~User.id.in_(session.query(User.id).filter_by(name='barry099'))).all()
print(res)

4.6 and or条件
from sqlalchemy import and_, or_

-or_包裹的都是or条件，and_包裹的都是and条件
res = session.query(User).filter(and_(User.id >= 3, User.name == 'barry099')).all()  # and条件
res = session.query(User).filter(or_(User.id < 2, User.name == 'barry099')).all()
res = session.query(User).filter(
    or_(
        User.id < 2,
        and_(User.name == 'barry099', User.id > 3),
        User.extra != ""
    )).all()
print(len(res))

4.7 通配符，以e开头，不以e开头
res = session.query(User).filter(User.email.like('%@%')).all()
res = session.query(User.id).filter(~User.name.like('e%'))
print(res)

4.8 分页
-一页2条，查第5页
res = session.query(User)[2 * 5:2 * 5 + 2]

4.9 排序，根据name降序排列（从大到小）
res = session.query(User).order_by(User.email.desc()).all()
res = session.query(Book).order_by(Book.price.desc()).all()
res = session.query(Book).order_by(Book.price.asc()).all()
# 第一个条件重复后，再按第二个条件升序排
ret = session.query(User).order_by(User.name.desc(), User.id.asc()).all()

5 分组查询
from sqlalchemy.sql import func

# res = session.query(User).group_by(User.extra).all()
# 分组之后取最大id，id之和，最小id
res = session.query(
    func.max(User.id),
    func.sum(User.id),
    func.min(User.id)).group_by(User.extra).all()
print(res)
for item in res:
    print(item[2])

-having
res = session.query(
    func.max(User.id),
    func.sum(User.id),
    func.min(User.id)).group_by(User.extra).having(func.max(User.id) > 2).all()
print(res)

5.1 链表操作
# select * from person,hobby where person.hobby_id=hobby.id;
res = session.query(Person, Hobby).filter(Person.hobby_id == Hobby.id).all()
-join表，默认是inner join,自动按外键关联
# select * from Person inner Hobby on Person.hobby_id=Hobby.id;
res = session.query(Person).join(Hobby).all()
-isouter=True 外连，表示Person left join Favor，没有右连接，反过来即可
# select * from Person left Hobby on Person.hobby_id=Hobby.id;
res = session.query(Person).join(Hobby, isouter=True).all()
-自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_,同上
res = session.query(Person).join(Hobby, Person.id == Hobby.id, isouter=True)  # sql本身有问题，自己指定链接字段
-右链接
res = session.query(Hobby).join(Person, isouter=True)
print(res)

6 组合UNION 操作符用于合并两个或多个 SELECT 语句的结果集
# union和union all的区别？
q1 = session.query(User).filter(User.id > 20)
q2 = session.query(User).filter(User.id > 18)
res = q1.union(q2).all()

q1 = session.query(User.email).filter(User.id > 20)
q2 = session.query(User.email).filter(User.id > 18)
res = q1.union_all(q2).all()

print(len(res))


7 一对多，基于链表跨表查（__链表）
# 方式一：直接连
res = session.query(Person, Hobby).filter(Person.hobby_id == Hobby.id, Hobby.id >= 2).all()
# 方式二：join连
res = session.query(Person).join(Hobby).filter(Person.id >= 2).all()

8 多对多关系，基于链表的跨表查
# 方式一：直接连
res = session.query(Boy, Girl, Boy2Girl).filter(Boy.id == Boy2Girl.boy_id, Girl.id == Boy2Girl.girl_id).all()
# 方式二：join连
res = session.query(Boy).join(Boy2Girl).join(Girl).filter(Person.id >= 2).all()

session.commit()
session.close()

flask-sqlalchemy、flask-migrate使用

1.flask中使用sqlalchemy，直接使用

1.1使用flask-sqlalchemy集成
-1 导入
	from flask_sqlalchemy import SQLAlchemy
-2 实例化得到对象
	db = SQLAlchemy()
-3 将db注册到app中
	db.init_app(app)
-4 视图函数中使用session
	全局的db.session  # 线程安全的
-5 models.py 中继承Base
	db.Base
-6 写字段 
	username = db.Column(db.String(80), unique=True, nullable=False)

2.flask-migrate
# python manage.py makemigrations  # 记录变化
# python manage.py migrate  # 把变化同步到数据库

2.1使用步骤：
-1 导入
	from flask_script import Manager
	from flask_migrate import Migrate, MigrateCommand
-2 注册
	manager = Manager(app)
	# 使用flask_migrate的Migrate  包裹一下app和db（sqlalchemy对象）
	Migrate(app, db)
-3 给flask_script增加一个db命令
	# 把命令增加到flask-script中去
	manager.add_command('db', MigrateCommand)
-4 出现3条命令
    python manage.py db init  # 只执行一次，做初始化操作，以后再也不执行了，多出一个migrations文件夹
    python manage.py db migrate  # 等同于django 的makemigrations
    python manage.py db upgrade  # 等同于django 的migrate