1. Johngo学长首页
  2. 技术点详细复盘专题
  3. Python

SQLAlchemy 从入门到入门

• Python • 阅读 72

1. SQLAlchemy 介绍

SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上，使用关系对象映射进行数据库操作，简言之便是：将类和对象转换成SQL，然后使用数据API执行SQL并获取执行结果。

安装： pip3 install sqlalchemy

组成部分：

Engine                      框架的引擎
Connection Pooling          数据库连接池
Dialect                     选择连接数据库的DB API种类
Schema/Types                架构和类型
SQL Exprression Language    SQL表达式语言

SQLAlchemy本身无法操作数据库，其必须以来 pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作

'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

django中如何反向生成models

python manage.py inspectdb > app/models.py

2. 简单使用（能创建表，删除表，不能修改表）

import threading
from sqlalchemy import create_engine

engine = create_engine(
    "mysql+pymysql://root:123@127.0.0.1:3306/d8?charset=utf8mb4",
    max_overflow=0,
    pool_size=5,
    pool_timeout=30,
    pool_recycle=-1
)

def task(arg):

    conn = engine.raw_connection()
    cursor = conn.cursor()
    cursor.execute(
        "select * from t8"
    )
    result = cursor.fetchall()
    print(result)
    cursor.close()
    conn.close()

for i in range(20):
    t = threading.Thread(target=task, args=(i,))
    t.start()
  • 模型表
  • orm 不能创建数据库，需要手动创建
  • sqlalchemy 只能创建表和删除表，不能新增，删除字段（使用第三方）
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()

class Users(Base):
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
    email = Column(String(32), unique=True)

    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'),
    )

def init_db():
"""
    根据类创建数据库表
    :return:
"""
    engine = create_engine(
        "mysql+pymysql://root:xwx@127.0.0.1:3306/flask_sql?charset=utf8mb4",
        max_overflow=0,
        pool_size=5,
        pool_timeout=30,
        pool_recycle=-1
    )

    Base.metadata.create_all(engine)

def drop_db():
"""
    根据类删除数据库表
"""
    engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/flask_sql?charset=utf8mb4",
        max_overflow=0,
        pool_size=5,
        pool_timeout=30,
        pool_recycle=-1
    )

    Base.metadata.drop_all(engine)

if __name__ == '__main__':

    init_db()
  • app.py（线程安全）
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
from sqlalchemy.orm import scoped_session
from threading import Thread

engine = create_engine(
    "mysql+pymysql://root:123@127.0.0.1:3306/flask_sql?charset=utf8mb4",
    max_overflow=0,
    pool_size=5,
    pool_timeout=30,
    pool_recycle=-1
)

Session = sessionmaker(bind=engine)

session = scoped_session(Session)

def task(i):

    user = Users(name='xwx_%s' % i, email='123%s@qq.com' % i, extra='hhh')
    session.add(user)

    session.commit()

    session.close()

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

3. 基本增删改查

...

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

1. 增加数据一条
session.add(对象)

2. 增加多条数据
session.add_all([对象A, 对象B])

res = session.query(Users).filter(Users.id == 1).delete()
print(res)

1. 查询后修改
res = session.query(Users).filter(Users.id > 0).update({"name": "xwx"})
print(res)

2. 类似于django的F查询，如果是字符类型设置synchronize_session为False追加字符串。数字类型设置为evaluate进行数值操作
session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "xwx"}, synchronize_session=False)
session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")

1. 查询所有
res = session.query(Users).all()
print(res)

2. 查询指定的字段类似于原生sql语句:  'select name as xx,age from users;'
res = session.query(Users.name.label('xx'), Users.age).all()
print(res)

1. filter 条件过滤，三种基本条件( > < == ) 且支持多条件连用
res = session.query(Users).filter(Users.id > 0, Users.name == 'xwx').all()
res1 = session.query(Users).filter(Users.name == 'xwx').first()

2. filter_by 条件过滤，使用表达式的形式（没有大于小于，且直接使用字段名过滤即可）
res = session.query(Users).filter_by(name='xwx', email='111@qq.com').all()

3. 自定义筛选条件
session.query(Users).filter(text("id>:vid or name=:vname")).params(vid=1, vname='xwx').all()
session.query(Users).filter(text("id>:vid and name=:vname")).params(vid=1, vname='xwx').all()

4. 自定义SQL语句
res = session.query(Users).from_statement(text("SELECT * FROM users where name='xwx'")).all()
res = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='xwx').all()

5. 二次筛选
session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='xwx'))).all()

1. and
session.query(Users).filter(Users.id > 1, Users.name == 'lqz').all()

2. between
session.query(Users).filter(Users.id.between(7, 9)).all()

3. in_
session.query(Users).filter(Users.id.in_([7, 8, 9])).all()

4. ~ 取反
session.query(Users).filter(~Users.id.in_([1, 3, 4])).all()

5. or_ 或
session.query(Users).filter(or_(Users.id > 2, Users.name == 'xwx')).all()

6. and_ 与
session.query(Users).filter(and_(Users.id > 3, Users.name == 'xwx')).all()

7. and_、or_ 连用
session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'xwx', Users.id > 3),
        Users.extra != ""
    ))


session.query(Users).filter(Users.name.like('xwx%')).all()
session.query(Users).filter(~Users.name.like('%xwx%')).all()

session.query(Users)[1:2]

1. 根据 age 降序排列（从大到小）desc()
session.query(Users).order_by(Users.age.desc()).all()

2. 根据 age 降序排列（从小到大）asc()
session.query(Users).order_by(Users.id.desc()).all()
session.query(Users).order_by(Users.id.asc()).all()

3. 第一个条件重复后，再按第二个条件升序排列
session.query(Users).order_by(Users.age.desc(), Users.id.asc()).all()

1. 按照 name 分组，等同于 sql 语句: 'select * from user group by name;'
session.query(Users).group_by(Users.name).all()

2. 分组后取最大、小值、总和，等同于: 'select max(id),sum(id),min(id) from user group by name;'
session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id),
).group_by(Users.name).all()

3. haviing 筛选，等同于: 'select max(id),sum(id),min(id) from user where email like 33% group by name  having min(id) > 2;'
session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)
).filter(Users.email.like('1%')).group_by(Users.name).having(func.min(Users.id) > 2).all()

1. 先笛卡尔积再过滤，等同于: 'select * from person,favor where user.id= favor.nid;'
session.query(Users, Favor).filter(Users.id == Favor.nid).all()

2. join 表，默认是 inner join，等同于: 'select * from person inner join favor on person.favor_id=favor.id;'
session.query(Person).join(Favor).all()

3. 左连接，添加isouter=True，没有右连接。'select * from person left join favor on person.favor_id=favor.id;'
session.query(Person).join(Favor, isouter=True).all()

4. 右连接，表反过来
session.query(Favor).join(Person, isouter=True).all()

5. 指定连接条件，类似于: 'select * from person left join favor on person.id=favor.id;'
session.query(Person).join(Favor, Person.id==Favor.id, isouter=True).all()

UNION：用于合并两个或多个 SELECT 语句的结果集，'不包括重复行'，同时进行默认规则的排序
Union All：对两个结果集进行并集操作，'包括重复行'，不进行排序；

1. union
q1 = session.query(Users.id).filter(Users.id > 3)
q2 = session.query(Users.id).filter(Users.id < 6)
ret = q1.union(q2).all()
print(ret)

2. union all
q1 = session.query(Users.id).filter(Users.id > 3)
q2 = session.query(Users.id).filter(Users.id < 6)
ret = q1.union_all(q2).all()

4 一对多外键关系

  • 表结构
    这里使用人与爱好表示例，假设一个人有一个爱好，爱好可以有多个人喜欢
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
from sqlalchemy.orm import relationship

Base = declarative_base()

class Hobby(Base):
    __tablename__ = 'hobby'
    id = Column(Integer, primary_key=True)
    caption = Column(String(50), default='篮球')

    def __repr__(self):
        return self.caption

class Person(Base):
    __tablename__ = 'person'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)

    hobby_id = Column(Integer, ForeignKey("hobby.id"))

    hobby = relationship('Hobby', backref='pers')

    def __repr__(self):
        return self.name

def init_db():
"""
    根据类创建数据库表
    :return:
"""
    engine = create_engine(
        "mysql+pymysql://root:xwx668428@127.0.0.1:3306/flask_sql?charset=utf8mb4",
        max_overflow=0,
        pool_size=5,
        pool_timeout=30,
        pool_recycle=-1
    )

    Base.metadata.create_all(engine)

if __name__ == '__main__':

    init_db()

1. 方式一
session.add(Hobby(caption='足球'))
session.add(Person(name='xwx', hobby_id=1))

2. 方式二，对象的方式
session.add(Person(name='xwx', hobby=Hobby(caption='橄榄球')))
  • 正向查询，根据人物查询爱好
res = session.query(Person).filter(Person.name == 'xwx').first()
print(res.hobby.caption)
  • 反向查询，根据爱好查询人物
res = session.query(Hobby).filter_by(caption='足球').first()
print(res.pers)
print(res.pers[0].id)

'select * from person，hobby where person.hobby_id=hobby.id and person.name=xwx;'
res = session.query(Person, Hobby).filter(Person.hobby_id == Hobby.id, Person.name == 'xwx').all()
print(res)
print(res[0])

'select * from person inner join hobby on person.hobby_id=hobby.id where person.name=xwx;'
res = session.query(Person, Hobby).join(Hobby).filter(Person.name == 'xwx').all()
print(res)

5. 多对多外键关系

  • *表模型
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 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 __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)

    girls = relationship('Girl', secondary='boy2girl', backref='boys')

    def __repr__(self):
        return self.name

...


1. 方式一，直接添加
session.add(Boy(name='B1'))
session.add(Girl(name='G1'))

session.add_all([Boy(name='B1'), Girl(name='G1')])

session.add_all([Boy2Girl(boy_id=1, girl_id=1), Boy2Girl(boy_id=1, girl_id=2)])

2. 方式二，使用外键
session.add(Boy(name='B1', girls=[Girl(name='G1'), Girl(name='G2')]))
  • *正向查询
res = session.query(Boy).filter_by(name='B1').first()
print(res.girls)
  • *反向查询
res = session.query(Girl).filter_by(name='G1').first()
print(res.boys)

6. flask-sqlalchemy、flask-migrate

  • flask-sqlalchemy 是 flask 和 SQLAchemy 的管理者，可以帮助我们快速把 sqlalchemy 集成到 flask 中。
  • flask-migrate ：原本 SQLAlchemy 不支持表字段增加删除。使用了 flask-migrate，以后增加了字段，删除字段只需要两条迁移命令就完成了
python3 manage.py db init       初始化，只执行一次
python3 manage.py db migrate    等同于 makemigartions
python3 manage.py db upgrade    等同于 migrate
  • *项目同名包/init .py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

from .models import *
from .views import account

def create_app():
    app = Flask(__name__)

    app.config.from_object('settings.DevelopmentConfig')

    db.init_app(app)

    app.register_blueprint(account.account)

    return app
  • *项目同名文件/models.py
from . import db

class Users(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)

    def __repr__(self):
        return '' % self.username
  • *项目同名文件/views/account.py
from flask import Blueprint
from .. import db
from .. import models

account = Blueprint('account', __name__)

@account.route('/login')
def login():

    db.session.add(models.Users(username='lqz', email='123'))
    db.session.query(models.Users).all()
    db.session.commit()
    return 'login'

Original: https://blog.csdn.net/m0_58987515/article/details/126289122
Author: XWenXiang
Title: SQLAlchemy 从入门到入门

原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/749074/

转载文章受原作者版权保护。转载请注明原作者出处！

赞 (0)
0

【自取】最近整理的，有需要可以领取学习：



Linux核心资料大放送~

全栈面试题汇总（持续更新&可下载）

一个提高学习100%效率的工具！

【超详细】深度学习面试题目！

LeetCode Python刷题答案下载！

LeetCode Java版刷题答案下载！

LeetCode C++ 版本，抓紧保存！

LeetCode GO语言 刷题答案下载！

大家都在看

亲爱的 Coder【最近整理，可免费获取】👉 最新必读书单  | 👏 面试题下载  | 🌎 免费的AI知识星球