sqlalchemy, конфигурация подключения flask-sqlalchemy и сведения об использовании

Flask

attention

использоватьsession.queryзаменятьmodel.query,model.queryДанные базы данных должны быть сопоставлены с моделью, и используются данные базы данных, которые должны быть изменены на лету.model.queryнельзя напрямую спрашивать

session.add()После выполнения оператор SQL будет добавлен к ожидающим и использован снова.session.queryБудет принудительно выполнять ожидающий SQL, а затем выполнять операцию запроса иmodel.querySQL в ожидании не будет применяться. (Конкретные принципы и методы реализации подлежат изучению.)

Журнал выглядит следующим образом:

>>> from app import db
>>> session = db.session
>>> from app import User
>>> user = User(name='test1', pwd='test1')
>>> session.add(user)
>>> session.query(User).filter()
>>> session.query(User).filter().all()
2022-01-06 16:57:54,286 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2022-01-06 16:57:54,287 INFO sqlalchemy.engine.Engine INSERT INTO user (name, pwd, addtime) VALUES (%(name)s, %(pwd)s, %(addtime)s)
2022-01-06 16:57:54,287 INFO sqlalchemy.engine.Engine [generated in 0.00019s] {'name': 'test1', 'pwd': 'test1', 'addtime': None}
2022-01-06 16:57:54,292 INFO sqlalchemy.engine.Engine SELECT user.id AS user_id, user.name AS user_name, user.pwd AS user_pwd, user.addtime AS user_addtime 
FROM user
2022-01-06 16:57:54,292 INFO sqlalchemy.engine.Engine [generated in 0.00038s] {}

session

Сначала создайте объект db, этот объект имеет всю инкапсуляцию базы данных в flask_sqlalchemy.

from flask import Flask
from flask_sqlalchemy imort SQLAlchemy


app = Flask(__name__)
db = SQLAlchemy(app, engine_options={'pool_recycle': 3600, 'pool_size': 20})
# session = db.session

Flask_sqlalchemy инкапсулирует многие операции SQLAlchemy. Вы можете напрямую использовать конфигурацию приложения в flask для создания соединений с базой данных. Вы можете использовать собственные методы SQLlchemy для создания соединений с базой данных. Механизмом sqlalchemy по умолчанию является MySQLdb.pip install mysql-python, но эта библиотека больше не поддерживает python3, она появится при установке в среде python3ModuleNotFoundError: No module named 'ConfigParser'ошибка, потому что python3 будетConfigParser.pyфайл изменен наconfigparser.py, скопируйте файл, то есть модуль, какConfigParser.pyможет решить проблему.

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
# 数据库+引擎://用户:密码@地址:端口/库?charset=utf8
mysql_uri = "mysql+pymysql://root:root@127.0.0.1:3306/SQLAlchemyTest?charset=utf8"

# paramstyle in ["qmark", "numeric", "named", "format" or "pyformat"]
engine = create_engine(
    mysql_uri,
    paramstyle=None,
    echo=True,
    echo_pool=True,
    pool_pre_ping=True,  # 悲观方式，每次执行语句先执行select语句实现ping的功能，可以在执行失败后重试
)
session_caller = sessionmaker(bind=engine, autoflush=True, autocommit=False, expire_on_commit=True, info=None)
session = session_caller()
# session = scoped_session(session_factory=Session_caller, scopefunc=None)

create_engineПринятые параметрыparamstyleявляется типом стиля параметра по умолчанию для python, вPEP249(paramstyle)подробно в

Обычно в качестве механизма sqlalchemy используется pymysql, но в случае без установления соединенияparamstyleНаступил на яму sqlalchemy и pymysql, pymysql находится в__init__.pyУказанный параметр ставится какpyformat, поэтому, если вы хотите указатьparamstyleЭтот параметр должен быть указан какNoneилиpyformat, иначе произойдет ошибка.

В режиме подключения работает нормально.

Если транзакция не задействована, рекомендуется режим соединения. то есть черезengine.connect()Создать соединение, выполнить SQL

использоватьnamed

# 使用named
engine = create_engine(mysql_uri, paramstyle='named')
session = sessionmaker(engine)()
sql = 'select * from dev_user where id={}'
session.execute(sql.format(':user_id'), {'user_id': 100}).fetchone()
# sqlalchemy 并没有将:user_id格式化，而是直接传给了pymysql，导致出错
# ProgrammingError: (pymysql.err.ProgrammingError) (1064, "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near ':user_id' at line 1")
# [SQL: select * from dev_user where id=:user_id]
# [parameters: {'user_id': 100}]

использоватьpyformat

# 使用pyformat
engine.paramstyle = 'pyformat'
session.execute(sql.format('%(user_id)s'), {'user_id': 100}).fetchone()
# sqlalchemy 没有获取到params参数
# ProgrammingError: (pymysql.err.ProgrammingError) (1064, "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near '%(user_id)s' at line 1")
# [SQL: select * from dev_user where id=%%(user_id)s]

# 而使用named风格的参数则没问题
session.execute(sql.format(':user_id'), {'user_id': 100}).fetchone()
# SQL:
# select * from dev_user where id=%(user_id)s
# {'user_id': 101}

Существует три способа использования sqlalchemy:

  • raw sql
  • sql expression
  • ORM

Первые два становятся основными методами,УведомлениеЕсли это sqlalchemy+cx_oracle, вам нужно отключить пул соединений, иначе будут исключения.Метод заключается в том, чтобы установить sqlalchemy.poolclass в sqlalchemy.pool.NullPool

поддержка транзакций

Создайте соединение, чтобы открыть транзакцию, и создайте точку сохранения для данных, которые добавляются без фиксации.

db.session.add(u1)
# 创建一个save point
db.session.begin_nested()

db.session.add(u2)
db.session.rollback()
# 只回退u2，不回退u1
db.session.commit()

Ниже приведена демонстрация модели базы данных.

class User(db.Model):
    __tablename__ = "dev_user"
    id = db.Column(INTEGER(unsigned=True), primary_key=True)
    username = db.Column(db.String(80), unique=True, doc="用户名", comment="用户id")
    email = db.Column(db.String(120), unique=True, doc="用户邮箱", comment="用户邮箱")
    is_active = db.Column(db.Boolean, default=False, doc="已激活", comment="是否已激活")
    phone = db.Column(db.String(20), index=True, doc="用户手机号", comment="用户手机号")
    age = db.Column(INTEGER(), doc="用户年龄", comment="用户年龄")

    def __repr__(self):
        return f"<User {self.username!r}>"


class Blog(db.Model):
    __tablename__ = "blog"
    id = db.Column(INTEGER(unsigned=True), primary_key=True)
    content = db.Column(db.Text, comment="博客内容")
    user_id = db.Column(INTEGER(unsigned=True), db.ForeignKey("dev_user.id"), index=True,
                        doc="用户id", comment="用户id")


class Comment(db.Model):
    __tablename__ = "comment"
    id = db.Column(INTEGER(unsigned=True), primary_key=True)
    comment = db.Column(db.String(150), doc="用户评论", comment="用户评论")
    user_id = db.Column(INTEGER(unsigned=True), index=True, doc"用户id", comment="用户id")


class UserInfo(db.Model):
    __tablename__ = "user_info"
    id = db.Column(INTEGER(unsigned=True), primary_key=True)
    user_id = db.Column(INTEGER(unsigned=True), db.ForeignKey("dev_user.id"),
												index=True, doc="用户id", comment="用户id",)
    user = db.relationship("User", backref="user_info", uselist=False)
    company = db.Column(db.String(100), unique=True, doc="公司简称", comment="公司简称")

Поддержка нескольких баз данных

flask_sqlalchemy использует концепцию привязки для привязки нескольких баз данных.

# 在配置config类中可以使用
class Config(object):
    SQLALCHEMY_DATABASE_URI = 'postgres://localhost/main'
    SQLALCHEMY_BINDS = {
    		'users':        'mysqldb://localhost/users',
    		'appmeta':      'sqlite:////path/to/appmeta.db'
		}
# 在app.config属性的设置中可以使用
app.config["SQLALCHEMY_DATABASE_URI"] = 'postgres://localhost/main'
app.config["SQLALCHEMY_BINDS"] = {
    'users':        'mysqldb://localhost/users',
    'appmeta':      'sqlite:////path/to/appmeta.db'
}

В flask_sqlalchemy есть много пакетов с параметрами привязки, которые могут указывать один из нескольких параметров базы данных, например

db.session.execute(sql, bind='user')

Разница в SQL:

-- 不使用bind
SELECT user.username AS username_1 FROM user WHERE user.id = 100;
-- 使用bind
SELECT test.user.username AS username_1 FROM test.user WHERE test.user.id = 100;
-- 表名前缀为数据库名

Использование поддержки нескольких баз данных в определении базы данных

class User(db.Model):
    __tablename__ = "user"
    # 指定`__bind_key__`在执行sql语句时会直接选择指定的数据库
    __bind_key__ = 'users'
    # 或者使用__table_args__
    __table_args__ = {
        'schema': "users",  # 指定bind
        'mysql_engine': 'InnoDB',
        'mysql_charset': 'utf8mb4',
    }
    # __table_args__也可以使用元组
    # __table_args__ = (
    #	    db.UniqueConstraint("username", "id", name="uix_username_id"),
    #     db.Index("ix_username", "username")
    # )
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True)
    # sqlalchemy的多态特性
    __mapper_args__ = {
        "order_by": username.desc()
    }

# 或者使用创建对象方式创建表模型
user_favorites = db.Table('user_favorites',
    db.Column('user_id', db.Integer, db.ForeignKey('user.id')),
    db.Column('message_id', db.Integer, db.ForeignKey('message.id')),
    info={'bind_key': 'users'}
)

Базовое использование (CRUD)

Простой запрос

  1. Добавить к

Используйте db.session для добавления данных

user = User(username='Jack', email='jack@email.com')
db.session.add(user)
db.session.commit()  # 手动commit
user = User(username='Vic', email='vic@email.com')
info = UserInfo(user=user, company='Google')
# 添加多个对象
db.session.add_all([user, info])
db,session.commit()
  1. удалять

удалить метод объекта запроса

Метод обновления и удаления принимает дополнительный параметрsynchronize_session=evaluate,synchronize_sessionДоступны три значения:

  • False: Обновите данные базы данных, не обновляйте данные, запрошенные в сеансе.
  • fetch: обновить данные базы данных и снова прочитать данные из базы данных.
  • evaluate: Параметр по умолчанию, если данные в сеансе не соответствуют данным в базе данных, сеанс не может определить, как с ним поступить, и он выкинетInvalidRequestErrorаномальный
rows = db.session.query(User).filter(User.id == 101).delete(synchronize_session='evaluate')

Однако, если вы удалите данные, связанные с внешним ключом напрямую, будет сообщено об ошибке, но вы можете использоватьsession.delete()Метод может обрабатывать конфигурацию внешнего ключа отношения.Если настроено cascade='all,delete-orphan', данные из таблицы будут удалены.

session.delete(instance)

db.session.query(User).filter(User.id == 101).delete()
# IntegrityError: (pymysql.err.IntegrityError) (1451, 'Cannot delete or update a parent row: a foreign key constraint fails (`test`.`user_info`, CONSTRAINT `user_info_ibfk_1` FOREIGN KEY (`user_id`) REFERENCES `dev_user` (`id`))')

instance = db.session.query(User).filter(User.id == 101).scalar()
db.session.delete(instance)
# 删除成功返回1
  1. Исправлять

Есть два способа изменить, изменить атрибут и затем отправить или изменить значение набора запросов.

# 修改对象属性
user = db.session.query(User).filter(User.id == 102).first()
user.username = 'Thomas'
db.session.add(user)
db.session.commit()
# 修改查询集Query的值
user_query = db.sessin.query(User.id < 100).update({'is_activate': 0})
db.session.commit()
  1. Запрос

filter_by

filter_byУсловие запроса=илиand, запрос с одним условием, условие должно иметь параметры ключевого слова, а соединение и возвращает объект запроса

# scalar只返回一个，如果没有或者有多个返回会抛出MultipleResultsFound异常
db.session.query(User).filter_by(id=101).scalar()
# 多个模型数据查询，默认查询第一个模型User的id为101的数据
db.session.query(User, UserInfo).filter_by(id=101).all()

filter

filterДопустимые типы параметров различаются, может быть несколько условий фильтрации, а возвращаемый объект запроса

В соответствии с идентификатором первичного ключа можно использовать напрямуюgetсделать запрос

db.session.query(User).get(100)
# 和Django一样，只获取一个结果，多个查询结果抛异常，没有则返回``None``

В отличие от Django ORM, sqlalchemy помещает некоторые условия в атрибуты поля модели, такие какorder_by,а такжеas

from sqlalchemy import desc, asc
query.order_by(desc(User.owned))
db.session.query(User.id).filter().order_by(asc(User.id)).offset(10).limit(10).all()
# SQL:
#  SELECT dev_user.id AS dev_user_id
# FROM dev_user ORDER BY dev_user.id DESC
# LIMIT %(param_1)s, %(param_2)s
# {'param_1': 10, 'param_2': 10}
db.session.query(User.username.label('name')).filter()
# 将username起别名为name
# SQL:
# SELECT dev_user.username as name
# FROM dev_user

Во-первых, это некоторые методы для получения данных из запроса,all(),first(),one(),scalar(),get(),one_or_none()

  • все: вернуть все данные

  • first: взять первую строку во всех строках, если результата нет, будет None, и исключение не будет выдано.

  • one: возвращаемые строки должны содержать только одну строку. Если строк несколько или данных нет, будет выдано исключение.

  • one_or_none: возвращаемые строки должны быть одной строкой или не содержать данных, исключение будет выдано, если есть несколько строк данных.

  • скаляр: возвращаемые строки должны быть одной строкой или не иметь данных.Если есть несколько строк данных, будет выдано исключение, и скаляр возьмет только первый результирующий набор в результате.

=, запрос

# from sqlalchemy import true
db.session.query(User.id, User.username).filter(User.is_activate==db.true())

# 组合查询
db.session.query(User.username, UserInfo.company).filter(User.id==103, User.username=='Zachary Moreno').all()
# 组合查询也可以使用and_，翻译成的SQL相同
from sqlalchemy import and_, or_
db.session.query(User.username, UserInfo.company).filter(and_(User.id==103, User.username=='Zachary Moreno')).all()
# 翻译成的SQL：
# SELECT dev_user.username AS dev_user_username, user_info.company AS user_info_company
# FROM dev_user, user_info
# WHERE dev_user.id = %(id_1)s AND dev_user.username = %(username_1)s
# params: {'id_1': 103, 'username_1': 'Zachary Moreno'}

db.session.query(User).filter(User.id < 20).all()
db.session.query(User).filter(User.create_at > '2020-02-02 00:00:00')

нечеткий запрос

db.session.query(User.username).filter(User.email.startswith('123')).all()
db.session.query(User.username).filter(User.email.like('123%'))
# SELECT dev_user.username from dev_user where dev_user.email like '123%'
db.session.query(User.username).filter(User.email.endswith('123')).all()

join

присоединиться к операции таблицы соединения

db.session.query(Comment.comment.label('user_comment'), User.id.label('user_id'), User.username.label('username')).join(User, User.id==Comment.user_id).filter(User.id<100).order_by(User.id.desc()).all()  # desc是字段的方法
# 对应SQL：
# SELECT comment.comment AS user_comment, dev_user.id AS user_id, dev_user.username AS username
# FROM comment INNER JOIN dev_user ON dev_user.id = comment.user_id
# WHERE dev_user.id = %(id_1)s
# params: {'id_1': 100}

Вы можете использовать псевдоним sqlalchemy для нескольких объединений.

from sqlalchemy.orm import aliased
Comment_alias = aliased(Comment, name='comment_alias')
db.session.query(User.id, User.username, User.email, User.phone, Comment.comment).join(Comment, Comment.user_id == User.id).join(Comment_alias, Comment_alias.user_id == User.id).all()
# SQL:
# SELECT dev_user.id AS dev_user_id, dev_user.username AS dev_user_username, dev_user.email AS dev_user_email, dev_user.phone AS dev_user_phone, comment.comment 
# FROM dev_user INNER JOIN comment ON comment.user_id = dev_user.id INNER JOIN comment AS comment_alias ON comment_alias.user_id = dev_user.id

offset/limit

from sqlalchemy import true

db.session.query(User.username, User.phone, User.email).filter(User.is_active == true()).order(User.age.desc()).offset(10).limit(20)
# SQL:
# SELECT dev_user.username AS dev_user_username, dev_user.phone AS dev_user_phone, dev_user.email AS dev_user_email
# FROM dev_user
# WHERE dev_user.is_active = true ORDER BY dev_user.age DESC
# LIMIT %(param_1)s, %(param_2)s
# {'param_1': 10, 'param_2': 20}

in/not in/is null/is not null

# in
db.session.query(User).filter(User.id.in_([100, 103, 105]))
# not in
db.session.query(User).filter(~User.id.in_([100, 103, 105]))
# is null
from sqlalchemy import null
db.session.query(User).filter(User.email == null())
# is not null
db.session.query(User).filter(User.email != null())

select_from

Укажите, из какой таблицы данных брать данные, ведь получаемые поля можно указать прямо в запросе.select_fromа такжеjoinТаблицы в не могут быть одинаковыми

db.session.query(User).select_from(Comment).join(User, Comment.user_id == User.id).all()
# 只获取User中的数据，即query中的model
# SELECT dev_user.id AS dev_user_id, dev_user.username AS dev_user_username, dev_user.email AS dev_user_email, dev_user.is_active AS dev_user_is_active, dev_user.phone AS dev_user_phone, dev_user.age AS dev_user_age
# FROM comment INNER JOIN dev_user ON dev_user.id = comment.user_id
Категории