介绍

Python SQLAlchemy 是一个强大且灵活的对象关系映射 (ORM) 库，可简化 Python 应用程序中的数据库交互。它为开发人员提供了一个高级的 Pythonic 接口来与关系数据库交互，允许他们使用 Python 类和对象来处理数据库，而不是编写复杂的 SQL 查询，并以这种方式抽象他们的代码，删除所有支持的数据库的任何本机语法系统，例如 PostgreSQL、MySQL、SQLite 和 Oracle。

范围

今天，我将向您演示 SQLAlchemy 库的三个有用功能，人们在任何类型的项目开发中每次使用该库时很可能都会用到这些功能。

尽管将解释该过程中包含的任何代码，但出于故事长度的考虑，将假定具有 Python 和 SQLAlchemy 或一般数据库的一些经验。如果您发现很难掌握，那么我建议您返回文档或大量指南来理解缺少的概念，然后返回重新阅读以获得更深入的理解。

使用案例

数据模型

对于今天的用例，我们将检查一个保存用户条目的简单 SQL 表。该代码将重点关注 SQLAlchemy 集成，以实现我们对数据存储的需求，因此不会涉及 Web 框架或任何其他复杂性。

为该示例定义的基本用户模型如下：

from pydantic import BaseModel
from typing import Optional


class UserSignUp(BaseModel):
    name: str
    surname: Optional[str] = None
    birth_year: Optional[int] = None
    notes: Optional[str] = None

在上面的代码片段中，我们定义了一个 Pydantic 模型，它是我们想要在数据库中保存记录的数据的模型表示。 Pydantic 是 Python 中最新且最有前途的库，用于处理数据验证、结构和建模。如果不熟悉它，请确保在某个时候花一些时间。

贮存

出于存储目的，我们将使用最简单的 SQL 数据库系统之一，即 SQLite。为此，我们需要定义一个引擎来与文件系统中实际的 SQLite 数据库文件进行通信，并创建一个会话对象，以便我们可以与数据库进行交互以进行操作。

from datetime import datetime
from pydantic import BaseModel
import os
import json
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
from sqlalchemy.sql import func
from sqlalchemy import Column, DateTime, Integer, String


project_dir = os.path.dirname(os.path.abspath(__file__))
engine = create_engine("sqlite:///{}".format(os.path.join(project_dir, "./storage.db")),
                       connect_args={"check_same_thread": False})
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=True)
    surname = Column(String, nullable=True)
    birth_year = Column(Integer, nullable=True)
    notes = Column(String, nullable=True)
    register_date = Column(DateTime, default=func.now())

    def __repr__(self):
        dict_repr = {c.name: getattr(self, c.name) for c in self.__table__.columns}
        for key, value in dict_repr.items():
            if isinstance(value, datetime):
                dict_repr[key] = datetime.isoformat(value)
        return json.dumps(dict_repr, indent=2)

Base.metadata.create_all(bind=engine)

在上面的代码片段中，项目文件夹路径用于指向数据库路径。数据库名称定义为 storage.db，如果尚不存在，则会自动创建。

此外，我们使用 SQLAlchemy ORM 模块的声明性基函数来构造一个基类，然后由我们的 User 类使用，该类代表数据库中将保存用户记录的表。

表用户行被定义为包含比我们为用户注册引入的数据模型额外的两个字段。

一个自动增量整数，作为已索引并用作表主键的行的 ID。

注册日期字段，它采用当前时间戳作为默认参数，这意味着正在数据库中写入的条目的时间戳。

数据库表定义类还包含一个额外的魔术方法 __repr__ ，它在此处定义以实现自定义序列化器。这样做的目的是我们可以直接打印数据库记录，即行，因为它们是使用 SQLAlchemy 库从表中获取的。如果这太多了，你可以忽略它，因为它的存在只是出于代表性的原因。

最后，通过调用Base.metadata模块的create_all方法，定义的表结构在我们创建的数据库中实例化。因此，列及其数据类型和主键定义现在都存在。

写入数据库表

既然我们定义并创建了数据库表，那么让我们尝试写入一条记录。为此，我们需要首先使用我们之前创建的引擎对象创建一个到数据库的会话。

from sqlalchemy import create_engine
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import sessionmaker


session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db_session = session_local()

然后，让我们使用声明的数据模型创建一个要注册的用户。

user_sign_up = UserSignUp(
    name="chris",
    surname="karvouniaris",
    notes="some notes about me"
)

最后，让我们使用用户的数据库表类创建一个条目，以使用我们建立的会话写入数据库并提交更改。

try:
    user_to_store = User(
        name=user_sign_up.name,
        surname=user_sign_up.surname,
        birth_year=user_sign_up.birth_year,
        notes=user_sign_up.notes
    )
    db_session.add(user_to_store)
    db_session.commit()
    print(user_to_store)
except IntegrityError as e:
    db_session.rollback()
    print(f"Error during user sign up, details: {str(e)}")

如果没有因任何完整性错误引发异常，则应写入该条目并将其提交到数据库。

该打印语句如下所示：

?  example python3.11 sqlalchemy_tricks.py
{
  "id": 17,
  "name": "chris",
  "surname": "karvouniaris",
  "birth_year": null,
  "notes": "some notes about me",
  "register_date": "2023-08-24T14:29:50"
}

请注意，birth_year 字段为空，因为我们首先没有为其提供任何整数，并且在写入操作期间会自动为 register_date 分配时间戳。

好的！现在我们已经了解了所有基础知识，让我们看看承诺的技巧！

1. 入库前验证和预处理

在许多用例中，在存储数据条目之前，我们首先需要预先验证或处理数据。

让我们看看我们将如何做到这一点。例如，假设我们想要将每个字符串字段大写。如果您需要在数据库中对数据进行标准化，这样的过程将非常有用。这种方法的一个示例好处是始终知道在搜索操作中要搜索的内容，而不必担心区分大小写。

那么，让我们看看如何在 SQLAlchemy 中使用其 ORM 验证装饰器来完成此操作。我们的表类 Users 应该重新调整以添加验证过程。

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=True)
    surname = Column(String, nullable=True)
    birth_year = Column(Integer, nullable=True)
    notes = Column(String, nullable=True)
    register_date = Column(DateTime, default=func.now())

    @validates('name', 'surname', 'notes')
    def normalize(self, _, value):
        if type(value) == str:
            return value.upper()

    def __repr__(self):
        dict_repr = {c.name: getattr(self, c.name) for c in self.__table__.columns}
        for key, value in dict_repr.items():
            if isinstance(value, datetime):
                dict_repr[key] = datetime.isoformat(value)
        return json.dumps(dict_repr, indent=2)

请注意，验证装饰器接受我们定义为字符串数据类型列的所有三列的参数。当然，这不是强制性的，人们可以使用任意数量的列并选择任何其他不同的操作。该进程的处理函数也很灵活，可以具有您喜欢的任何名称及其功能。

我们的规范化函数将应用于所有选定的三列，这意味着传入数据将通过此功能并以我们在写入数据库之前需要的处理格式返回。在我们的例子中，一个简单的验证就足够了，即如果传递的值是字符串类型，则应将其大写。

我们之前运行的相同写入操作现在将打印以下内容：

?  example python3.11 sqlalchemy_tricks.py
{
  "id": 17,
  "name": "CHRIS",
  "surname": "KARVOUNIARIS",
  "birth_year": null,
  "notes": "SOME NOTES ABOUT ME",
  "register_date": "2023-08-24T15:29:50"
}

2. 用于动态查询数据的列属性

在 Python SQLAlchemy 中，column_property 函数用于在 SQLAlchemy 类上创建计算属性或别名属性，该属性通常源自数据库表中的一列或多列。此函数允许您在 SQLAlchemy 模型上定义其他属性，这些属性不直接映射到数据库中的单个列，而是从现有列计算或派生。

让我们探讨一个示例，我们将根据用户的出生年份得出用户的年龄。

Users 表类将以这种方式重新调整。

from sqlalchemy import Column, DateTime, Integer, String
from sqlalchemy.sql import func, extract
from sqlalchemy.orm import column_property, validates


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=True)
    surname = Column(String, nullable=True)
    birth_year = Column(Integer, nullable=True)
    age = column_property(extract('year', func.now()) - birth_year)
    notes = Column(String, nullable=True)
    register_date = Column(DateTime, default=func.now())

    @validates('name', 'surname', 'notes')
    def normalize(self, _, value):
        if type(value) == str:
            return value.upper()

    def __repr__(self):
        dict_repr = {c.name: getattr(self, c.name) for c in self.__table__.columns}
        for key, value in dict_repr.items():
            if isinstance(value, datetime):
                dict_repr[key] = datetime.isoformat(value)
        dict_repr['age'] = self.age
        return json.dumps(dict_repr, indent=2)

请注意我们的类定义中的新字段 Age。它不是列类型，而是包含一个短表达式。使用来自 SQLAlchemy 的 sql 模块的 extract 方法，我们从由 func.now() 派生的时间戳中提取年份。我们以前见过这个。每当查询表并动态派生时间戳时，该函数就会运行。通过提取时间戳的年份字段，我们正在提取年份整数。最后减去用户的出生年份，就可以计算出用户的年龄。该值不会存储在表行中的任何位置，但会在返回行时返回，因为它是动态计算的。

由于这个年龄字段不是一个实际的列，它的值不会包含在我们迄今为止使用的 self.__table__.columns 字典中。还有其他方法可以获取该值，但为了本教程的目的，我们将通过将类实例属性作为值包含在返回的字典中手动添加此键值对。

让我们看看这次写入数据库会是什么样子：

user_sign_up = UserSignUp(
    name="chris",
    surname="karvouniaris",
    notes="some notes about me",
    birth_year=1992
)

try:
    user_to_store = User(
        name=user_sign_up.name,
        surname=user_sign_up.surname,
        birth_year=user_sign_up.birth_year,
        notes=user_sign_up.notes
    )
    db_session.add(user_to_store)
    db_session.commit()
    print(user_to_store)
except IntegrityError as e:
    db_session.rollback()
    print(f"Error during user sign up, details: {str(e)}")

这次，我们提供出生年份，以便执行计算。

输出如下所示：

?  example python3.11 sqlalchemy_tricks.py
{
  "id": 1,
  "name": "CHRIS",
  "surname": "KARVOUNIARIS",
  "birth_year": 1992,
  "notes": "SOME NOTES ABOUT ME",
  "register_date": "2023-08-24T15:58:31",
  "age": 31
}

瞧！现在，年龄字段会动态填充并在每次查询时返回。例如，如果我们在一年过去后查询这条完全相同的记录，则该记录将完全相同，但有效负载中的年龄字段将增加到 32 。

3、主键之上多列的唯一约束

任何熟悉 SQL 数据库系统的人从一开始就知道每个表都需要包含一个主键。主键是唯一标识表中每条记录（行）的基本概念。它确保指定列中不存在重复值或空值，从而保证数据的完整性和唯一性。

然而，有时，我们需要对记录条目进行更多限制。在某些用例中，主键并不能满足我们的所有需求。

让我们看一个用例中的例子。假设传入用户与现有用户具有完全相同的姓名。我们的表主键只是一个增量数字，因此允许输入。现在，为了便于论证，我们假设我们可以使用姓名字段以及出生年份字段来标识两个不同的记录。也就是说，如果现有用户的所有这三个字段都相同，则应拒绝条目。

让我们看看如何在 Users 表中强制执行这样的规则：

import json
from sqlalchemy import UniqueConstraint
from sqlalchemy import Column, DateTime, Integer, String
from sqlalchemy.sql import func, extract
from sqlalchemy.orm import column_property, validates




class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=True)
    surname = Column(String, nullable=True)
    birth_year = Column(Integer, nullable=True)
    age = column_property(extract('year', func.now()) - birth_year)
    notes = Column(String, nullable=True)
    register_date = Column(DateTime, default=func.now())
    __table_args__ = (UniqueConstraint('name', 'surname', 'birth_year', name='unique_fullname_per_birth_year'),)

    @validates('name', 'surname', 'notes')
    def normalize(self, _, value):
        if type(value) == str:
            return value.upper()

    def __repr__(self):
        dict_repr = {c.name: getattr(self, c.name) for c in self.__table__.columns}
        for key, value in dict_repr.items():
            if isinstance(value, datetime):
                dict_repr[key] = datetime.isoformat(value)
        dict_repr['age'] = self.age
        return json.dumps(dict_repr, indent=2)

请注意由 UniqueConstraint 对象提供的表参数。该对象采用不同的列名称作为参数，这些列名称应形成每个记录的唯一性。此外，您还可以为规则提供名称。

让我们看看如果运行与上一个示例中完全相同的脚本会发生什么：

user_sign_up = UserSignUp(
    name="chris",
    surname="karvouniaris",
    notes="some notes about me",
    birth_year=1992
)

try:
    user_to_store = User(
        name=user_sign_up.name,
        surname=user_sign_up.surname,
        birth_year=user_sign_up.birth_year,
        notes=user_sign_up.notes
    )
    db_session.add(user_to_store)
    db_session.commit()
    print(user_to_store)
except IntegrityError as e:
    db_session.rollback()
    print(f"Error during user sign up, details: {str(e)}")

输出将显示为：

?  example python3.11 sqlalchemy_tricks.py
Error during user sign up, details: (sqlite3.IntegrityError) UNIQUE constraint failed: users.name, users.surname, users.birth_year
[SQL: INSERT INTO users (name, surname, birth_year, notes, register_date) VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP) RETURNING id, register_date]
[parameters: ('CHRIS', 'KARVOUNIARIS', 1992, 'SOME NOTES ABOUT ME')]
(Background on this error at: https://sqlalche.me/e/20/gkpj)

请注意，尝试在数据库中提交更改会引发异常。提交尝试的更改将被回滚，并且条目将被阻止。

完整代码

添加所有内容后的完整代码如下所示：

from datetime import datetime
from pydantic import BaseModel
from typing import Optional
import os
import json
from sqlalchemy import create_engine, UniqueConstraint
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, DateTime, Integer, String
from sqlalchemy.sql import func, extract
from sqlalchemy.orm import column_property, validates



project_dir = os.path.dirname(os.path.abspath(__file__))
engine = create_engine("sqlite:///{}".format(os.path.join(project_dir, "./storage.db")),
                       connect_args={"check_same_thread": False})
Base = declarative_base()


class UserSignUp(BaseModel):
    name: str
    surname: Optional[str] = None
    birth_year: Optional[int] = None
    notes: Optional[str] = None


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=True)
    surname = Column(String, nullable=True)
    birth_year = Column(Integer, nullable=True)
    age = column_property(extract('year', func.now()) - birth_year)
    notes = Column(String, nullable=True)
    register_date = Column(DateTime, default=func.now())
    __table_args__ = (UniqueConstraint('name', 'surname', 'birth_year', name='unique_fullname_per_birth_year'),)

    @validates('name', 'surname', 'notes')
    def normalize(self, _, value):
        if type(value) == str:
            return value.upper()

    def __repr__(self):
        dict_repr = {c.name: getattr(self, c.name) for c in self.__table__.columns}
        for key, value in dict_repr.items():
            if isinstance(value, datetime):
                dict_repr[key] = datetime.isoformat(value)
        dict_repr['age'] = self.age
        return json.dumps(dict_repr, indent=2)

Base.metadata.create_all(bind=engine)

session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db_session = session_local()

user_sign_up = UserSignUp(
    name="chris",
    surname="karvouniaris",
    notes="some notes about me",
    birth_year=1992
)

try:
    user_to_store = User(
        name=user_sign_up.name,
        surname=user_sign_up.surname,
        birth_year=user_sign_up.birth_year,
        notes=user_sign_up.notes
    )
    db_session.add(user_to_store)
    db_session.commit()
    print(user_to_store)
except IntegrityError as e:
    db_session.rollback()
    print(f"Error during user sign up, details: {str(e)}")

结论

在处理任何规模的 Python 项目中的数据库时，SQLAlchemy 是一个必须了解的工具。这些只是探索的一些非常基本的功能，可能对出现的许多案例都很有用。

这就是大家！

Python 的 SQLAlchemy:您需要了解的 3 个强大功能

介绍

范围

使用案例

数据模型

贮存

写入数据库表

1. 入库前验证和预处理

2. 用于动态查询数据的列属性

3、主键之上多列的唯一约束

完整代码

结论

相关推荐

取消回复欢迎你发表评论:

生产系统数据库操作规范(2):（数据库生产库是什么意思）

Redis客户端 Jedis 与 Lettuce

高并发架构系列:Redis并发竞争key的解决方案详解

redis如何防止并发（redis如何防止高并发）

开源推荐:如何实现的一个高性能 Redis 服务器

redis安装与调优部署文档(WinServer)

Redis 入门 - 安装最全讲解(Windows、Linux、Docker)

一文带你了解 Redis 的发布与订阅的底层原理

Redis如何应对并发访问（redis控制并发量）

Java SE Development Kit 8u441下载地址【windows版本】

Python 的 SQLAlchemy:您需要了解的 3 个强大功能

介绍

范围

使用案例

数据模型

贮存

写入数据库表

1. 入库前验证和预处理

2. 用于动态查询数据的列属性

3、主键之上多列的唯一约束

完整代码

结论

相关推荐

取消回复欢迎 你 发表评论:

生产系统数据库操作规范(2):（数据库生产库是什么意思）

Redis客户端 Jedis 与 Lettuce

高并发架构系列:Redis并发竞争key的解决方案详解

redis如何防止并发（redis如何防止高并发）

开源推荐:如何实现的一个高性能 Redis 服务器

redis安装与调优部署文档(WinServer)

Redis 入门 - 安装最全讲解(Windows、Linux、Docker)

一文带你了解 Redis 的发布与订阅的底层原理

Redis如何应对并发访问（redis控制并发量）

Java SE Development Kit 8u441下载地址【windows版本】

取消回复欢迎你发表评论: