有勇气的牛排博客
☰

sqlalchemy1.4风格（2022新语法配合协程更优）

有勇气的牛排 1083 Python 2023-05-18 20:58:14
知止自律打卡-微信小程序APP
一款微信自律打卡小程序，支持自定义打卡项目、习惯记录、项目日记、专注计时与数据统计。自律打卡、学习打卡、读书打卡、健身打卡、运动打卡、减肥打卡、戒色打卡、养生打卡、戒烟、戒酒、戒熬夜等多种场景。万能打卡模板自由创建。可自定义每日打卡目标，记录习惯连续天数，可视化追踪成长进度。
了解产品详情 →
微信搜索「知止自律打卡」
知止自律打卡微信小程序

1 基本配置

官方文档：
https://docs.sqlalchemy.org/en/14/orm/session_basics.html#session-querying-20

1.1 Model

from sqlalchemy import INTEGER, Column, ForeignKey, String
from sqlalchemy.orm import declarative_base, relationship

Base = declarative_base()

class Users(Base):
    __tablename__ = "users"
    uid = Column(INTEGER, primary_key=True)
    name = Column(String(30))
    age = Column(INTEGER)

class Article(Base):
    __tablename__ = "article"
    a_id = Column(INTEGER, primary_key=True)
    uid = Column(INTEGER)
    a_title = Column(String(30))
    a_readcount = Column(INTEGER)

1.2 映射

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from model import Base

# 建立与MySQL的连接
db_name = "sanic_test"
host = "127.0.0.1"
db_user = "root"
password = "root"

engine = create_engine(f'mysql+pymysql://{db_user}:{password}@{host}/{db_name}', echo=True)

# 创建session对象
session = sessionmaker(engine)()

# 创建表，执行所有BaseModel类的子类
Base.metadata.create_all(engine)

# 提交，必须
session.commit()

1.3 main

from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session

from model import Users, Article

engine = create_engine('mysql://root:root@127.0.0.1/sanic_test?charset=utf8')
session = Session(engine, future=True)

2 数据库

mysql> select * from users;
+-----+--------------+-----+
| uid | name         | age |
+-----+--------------+-----+
|   1 | 有勇气的牛排 |  20 |
|   2 | 灰太狼       |  20 |
+-----+--------------+-----+

mysql> select * from article;
+------+-----+---------+-------------+
| a_id | uid | a_title | a_readcount |
+------+-----+---------+-------------+
|    1 |   1 | python  |          10 |
|    2 |   1 | java    |          30 |
|    3 |   2 | vue     |          15 |
+------+-----+---------+-------------+

3 异常处理

with Session(engine) as session:
    session.begin()
    try:
        session.add(some_object)
        session.add(some_other_object)
    except:
        session.rollback()
        raise
    else:
        session.commit()

结合写法

with Session(engine) as session, session.begin():
    session.add(some_object)
    session.add(some_other_object)

4 使用 sessionmaker

from sqlalchemy.orm import sessionmaker

engine = create_engine('postgresql://scott:tiger@localhost/')
# a sessionmaker(), also in the same scope as the engine
Session = sessionmaker(engine)

3 增加数据

没有的列，自动为空

user1 = Users(name="战神")
user2 = Users(name="战神2", age=18)
session.add(user1)
session.add(user2)

session.commit()

要一次将项目列表添加到会话中，可以使用 Session.add_all()

session.add_all([user1, user2])

4 删除

5 更新

stmt = update(Users).where(Users.uid == 7).values(name="大哥").\
    execution_options(synchronize_session="fetch")

result = session.execute(stmt)

execution_options()：可用于建立执行时间选项

5 查询

5.1 查询单表 多个记录

statement = select(Users).filter_by(age=20)
result = session.execute(statement).scalars().all()

print(result)
for i in result:
    print(i)
    print(i.id)
    print(i.name)
    print(i.age)

5.2 查询多表 多个记录

statement = select(Users, Article).where(Users.uid == Article.uid)
result = session.execute(statement).all()

print(result)
for users, article in result:
    print(users.uid, article.a_title)

5.3 指定列

statement = select(Users.uid, Users.name, Article).where(Users.uid == Article.uid)
result = session.execute(statement).all()

print(result)
for uid, name, article in result:
    print(uid, name, article.a_title)

标签：

×
×