使用 SQLAlchemy 和 Python 对象

SQLAlchemy是一个强大的 Python 数据库访问工具包，其对象关系映射器 (ORM)是其最著名的组件之一，此处讨论和使用了该组件。

当您使用面向对象的语言（如 Python）工作时，从对象的角度进行思考通常很有用。可以将 SQL 查询返回的结果映射到对象，但这样做与数据库的工作方式背道而驰。坚持使用 SQL 提供的标量结果与 Python 开发人员的工作方式背道而驰。这个问题被称为对象-关系阻抗失配。

SQLAlchemy 提供的 ORM 位于 SQLite 数据库和 Python 程序之间，并转换数据库引擎和 Python 对象之间的数据流。SQLAlchemy 允许您从对象的角度进行思考，同时仍保留数据库引擎的强大功能。

from sqlalchemy import Column, Integer, String, ForeignKey, Table
from sqlalchemy.orm import relationship, backref
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

author_publisher = Table(
    "author_publisher",
    Base.metadata,
    Column("author_id", Integer, ForeignKey("author.author_id")),
    Column("publisher_id", Integer, ForeignKey("publisher.publisher_id")),
)

book_publisher = Table(
    "book_publisher",
    Base.metadata,
    Column("book_id", Integer, ForeignKey("book.book_id")),
    Column("publisher_id", Integer, ForeignKey("publisher.publisher_id")),
)

class Author(Base):
    __tablename__ = "author"
    author_id = Column(Integer, primary_key=True)
    first_name = Column(String)
    last_name = Column(String)
    books = relationship("Book", backref=backref("author"))
    publishers = relationship(
        "Publisher", secondary=author_publisher, back_populates="authors"
    )

class Book(Base):
    __tablename__ = "book"
    book_id = Column(Integer, primary_key=True)
    author_id = Column(Integer, ForeignKey("author.author_id"))
    title = Column(String)
    publishers = relationship(
        "Publisher", secondary=book_publisher, back_populates="books"
    )

class Publisher(Base):
    __tablename__ = "publisher"
    publisher_id = Column(Integer, primary_key=True)
    name = Column(String)
    authors = relationship(
        "Author", secondary=author_publisher, back_populates="publishers"
    )
    books = relationship(
        "Book", secondary=book_publisher, back_populates="publishers"
    )

ForeignKey 创建连接

SQLAlchemyForeignKey类定义了Column不同表中两个字段之间的依赖关系。AForeignKey是如何让 SQLAlchemy 了解表之间的关系。例如，author_publisher实例创建中的这一行建立了外键关系：

Column("author_id", Integer, ForeignKey("author.author_id"))

books = relationship("Book", backref=backref("author"))

上面的代码定义了一个父子集合。该books属性存在复数（这不是必须的，只是一个约定）是一个迹象，这是一个集合。

第一个参数relationship()是类名Book（不是表名book），是与books属性相关的类。该relationship运筹学SQLAlchemy的有总之间的关系Author和Book阶级。SQLAlchemy 会在Book类定义中找到关系：

author_id = Column(Integer, ForeignKey("author.author_id"))

SQLAlchemy 认识到这是ForeignKey两个类之间的连接点。稍后您将获得backref参数relationship()。

中的另一个关系Author是与Publisher班级的关系。这是使用Author类定义中的以下语句创建的：

publishers = relationship(
    "Publisher", secondary=author_publisher, back_populates="authors"
)

book = session.query(Book).filter_by(Book.title == "The Stand").one_or_none()
print(f"Authors name: {book.author.first_name} {book.author.last_name}")

上面的author属性存在Book是因为backref定义。backref当您需要引用父级并且您拥有的只是一个子实例时，使用A可能非常方便。

查询回答问题

您可以像这样SELECT * FROM author;在 SQLAlchemy 中进行基本查询：

results = session.query(Author).all()

的session是用于在Python示例程序的SQLite通信的SQLAlchemy的对象。在这里，您告诉会话您要对Author模型执行查询并返回所有记录。

在这一点上，使用 SQLAlchemy 代替普通 SQL 的优势可能并不明显，尤其是考虑到创建代表数据库的模型所需的设置。该results查询返回的神奇在哪里发生。您将返回具有Author与您定义的列名称匹配的属性的对象实例列表，而不是返回标量数据列表的列表。

由 SQLAlchemy 维护的books和publishers集合创建了作者和他们所写书籍以及出版这些书籍的出版商的分层列表。

在幕后，SQLAlchemy 将对象和方法调用转换为 SQL 语句以针对 SQLite 数据库引擎执行。SQLAlchemy 将 SQL 查询返回的数据转换为 Python 对象。

使用 SQLAlchemy，您可以对作者列表和他们所写的书籍数量执行前面显示的更复杂的聚合查询，如下所示：

author_book_totals = (
    session.query(
        Author.first_name,
        Author.last_name,
        func.count(Book.title).label("book_total")
    )
    .join(Book)
    .group_by(Author.last_name)
    .order_by(desc("book_total"))
    .all()
)

def main():
    """Main entry point of program"""
    # Connect to the database using SQLAlchemy
    with resources.path(
        "project.data", "author_book_publisher.db"
    ) as sqlite_filepath:
        engine = create_engine(f"sqlite:///{sqlite_filepath}")
    Session = sessionmaker()
    Session.configure(bind=engine)
    session = Session()

    # Get the number of books printed by each publisher
    books_by_publisher = get_books_by_publishers(session, ascending=False)
    for row in books_by_publisher:
        print(f"Publisher: {row.name}, total books: {row.total_books}")
    print()

    # Get the number of authors each publisher publishes
    authors_by_publisher = get_authors_by_publishers(session)
    for row in authors_by_publisher:
        print(f"Publisher: {row.name}, total authors: {row.total_authors}")
    print()

    # Output hierarchical author data
    authors = get_authors(session)
    output_author_hierarchy(authors)

    # Add a new book
    add_new_book(
        session,
        author_name="Stephen King",
        book_title="The Stand",
        publisher_name="Random House",
    )
    # Output the updated hierarchical author data
    authors = get_authors(session)
    output_author_hierarchy(authors)

该程序是examples/example_1/main.py. 让我们来看看差异：

• 第 4 到 7 行首先将sqlite_filepath变量初始化为数据库文件路径。然后他们创建engine变量以与 SQLite 和author_book_publisher.db数据库文件进行通信，这是 SQLAlchemy 对数据库的访问点。

• 8号线创建Session从SQLAlchemy的的类sessionmaker()。

• 第 9Session行将绑定到第 8 行中创建的引擎。

• 第 10 行创建session实例，程序使用该实例与 SQLAlchemy 进行通信。

该函数的其余部分类似，只是将datawith替换session为所有被 调用的函数的第一个参数main()。

get_books_by_publisher() 已重构为使用 SQLAlchemy 和您之前定义的模型来获取请求的数据：

def get_books_by_publishers(session, ascending=True):
 2    """Get a list of publishers and the number of books they've published"""
 3    if not isinstance(ascending, bool):
 4        raise ValueError(f"Sorting value invalid: {ascending}")
 5
 6    direction = asc if ascending else desc
 7
 8    return (
 9        session.query(
10            Publisher.name, func.count(Book.title).label("total_books")
11        )
12        .join(Publisher.books)
13        .group_by(Publisher.name)
14        .order_by(direction("total_books"))
15    )

这是新函数get_books_by_publishers(), 正在执行的操作：

• 第 6 行创建direction变量并根据参数的值将其设置为等于 SQLAlchemydesc或asc函数ascending。

• 第 9 行到第 11 行查询Publisher要返回的数据表，在本例中为Publisher.name与Book作者相关联的对象总数，别名为total_books。

• 第 12 行连接到Publisher.books集合。

• 第 13 行按Publisher.name属性聚合图书计数。

• 第 14 行根据由 定义的运算符按图书计数对输出进行排序direction。

• 第 15 行关闭对象，执行查询，并将结果返回给调用者。

上面的所有代码都表达了想要的东西，而不是如何检索它。现在，您不再使用 SQL 来描述所需内容，而是使用 Python 对象和方法。返回的是 Python 对象列表，而不是数据元组列表。

get_authors_by_publisher()也被修改为专门与 SQLAlchemy 一起使用。它的功能与前面的功能非常相似，因此省略了功能说明：

def get_authors_by_publishers(session, ascending=True):
    """Get a list of publishers and the number of authors they've published"""
    if not isinstance(ascending, bool):
        raise ValueError(f"Sorting value invalid: {ascending}")

    direction = asc if ascending else desc

    return (
        session.query(
            Publisher.name,
            func.count(Author.first_name).label("total_authors"),
        )
        .join(Publisher.authors)
        .group_by(Publisher.name)
        .order_by(direction("total_authors"))
    )

get_authors()已添加以获取按姓氏排序的作者列表。此查询的结果是一个Author包含书籍集合的对象列表。该Author物体已经包含分层数据，所以结果不必重新格式化：

def get_authors(session):
    """Get a list of author objects sorted by last name"""
    return session.query(Author).order_by(Author.last_name).all()

与其之前的版本一样，add_new_book()相对复杂但易于理解。它确定数据库中是否已经存在具有相同标题、作者和出版商的书籍。

如果搜索查询找到完全匹配，则函数返回。如果没有书与确切的搜索条件匹配，则它会搜索以查看作者是否使用传入的书名写了一本书。此代码的存在是为了防止在数据库中创建重复的书籍。

如果不存在匹配的书籍，并且作者没有写过同名书籍，则会创建一本新书。然后该函数检索或创建作者和出版商。一旦Book,Author和 的实例Publisher存在，它们之间的关系就会被创建，并将结果信息保存到数据库中：

def add_new_book(session, author_name, book_title, publisher_name):
    """Adds a new book to the system"""
    # Get the author's first and last names
    first_name, _, last_name = author_name.partition(" ")

    # Check if book exists
    book = (
        session.query(Book)
        .join(Author)
        .filter(Book.title == book_title)
        .filter(
            and_(
                Author.first_name == first_name, Author.last_name == last_name
            )
        )
        .filter(Book.publishers.any(Publisher.name == publisher_name))
        .one_or_none()
    )
    # Does the book by the author and publisher already exist?
    if book is not None:
        return

    # Get the book by the author
    book = (
        session.query(Book)
        .join(Author)
        .filter(Book.title == book_title)
        .filter(
            and_(
                Author.first_name == first_name, Author.last_name == last_name
            )
        )
        .one_or_none()
    )
    # Create the new book if needed
    if book is None:
        book = Book(title=book_title)

    # Get the author
    author = (
        session.query(Author)
        .filter(
            and_(
                Author.first_name == first_name, Author.last_name == last_name
            )
        )
        .one_or_none()
    )
    # Do we need to create the author?
    if author is None:
        author = Author(first_name=first_name, last_name=last_name)
        session.add(author)

    # Get the publisher
    publisher = (
        session.query(Publisher)
        .filter(Publisher.name == publisher_name)
        .one_or_none()
    )
    # Do we need to create the publisher?
    if publisher is None:
        publisher = Publisher(name=publisher_name)
        session.add(publisher)

    # Initialize the book relationships
    book.author = author
    book.publishers.append(publisher)
    session.add(book)

    # Commit to the database
    session.commit()

上面的代码比较长。让我们将功能分解为可管理的部分：

• 第 7 到 18行将book变量设置为 a 的实例，Book如果找到具有相同标题、作者和出版商的书。否则，它们设置book为None。

• 第 20 和 21 行确定这本书是否已经存在，如果存在则返回。

• 如果找到具有相同标题和作者的书，第 24 到 37行将book变量设置为 a 的实例Book。否则，他们会创建一个新Book实例。

• 第 40 到 52行将author变量设置为现有作者（如果找到），或者Author根据传入的作者姓名创建一个新实例。

• 第 55 到 63行将publisher变量设置为现有发布者（如果找到），或者Publisher根据传入的发布者名称创建新实例。

• 第 66行将book.author实例设置为author实例。这会在作者和书籍之间创建关系，提交会话时 SQLAlchemy 将在数据库中创建该关系。

• 第 67行将publisher实例添加到book.publishers集合中。这在book和publisher表之间创建了多对多关系。SQLAlchemy 将在表以及book_publisher连接两者的关联表中创建引用。

• 第 68行将Book实例添加到会话中，使其成为会话工作单元的一部分。

• 第 71行将所有创建和更新提交到数据库。

这里有几点需要注意。首先，在查询或创建和更新中都没有提及author_publisher或book_publisher关联表。由于您在models.py设置关系方面所做的工作，SQLAlchemy 可以处理将对象连接在一起并在创建和更新期间保持这些表同步。

其次，所有的创建和更新都发生在session对象的上下文中。这些活动都没有触及数据库。只有当session.commit()语句执行时，会话才会通过其工作单元并将该工作提交给数据库。

例如，如果Book创建了一个新实例（如上面的第 37 行），那么除了book_id主键和author_id外键之外，书的属性都会被初始化。因为还没有发生数据库活动，所以book_id是未知的，并且在实例化中没有做任何事情book来给它一个author_id.

当session.commit()被执行时，的事情会做一个是插入book到数据库中，此时数据库将创建book_id主键。然后会话将book.book_id使用数据库引擎创建的主键值初始化该值。

session.commit()也知道Book实例在author.books集合中的插入。该author对象的author_id主键将被添加到Book附加到实例author.books集合作为author_id外键。

为多个用户提供访问权限
至此，您已经了解了如何使用 pandas、SQLite 和 SQLAlchemy 以不同的方式访问相同的数据。对于作者、书籍和出版商数据的相对简单的用例，是否应该使用数据库仍然是一个折腾。

在选择使用平面文件还是数据库时，一个决定性因素是数据和关系的复杂性。如果每个实体的数据很复杂并且包含实体之间的许多关系，那么在平面文件中创建和维护它可能会变得更加困难。

另一个需要考虑的因素是您是否要在多个用户之间共享数据。这个问题的解决方案可能就像使用运动鞋网在用户之间物理移动数据一样简单。以这种方式移动数据文件的优点是易于使用，但在进行更改时数据可能会很快失去同步。

如果用户是远程的并且想要跨网络访问数据，那么保持所有用户的数据一致的问题就变得更加困难。即使您仅限于使用 Python 等单一语言并使用 Pandas 访问数据，网络文件锁定也不足以确保数据不会被损坏。

通过服务器应用程序和用户界面提供数据可以缓解这个问题。服务器是唯一需要对数据库进行文件级访问的应用程序。通过使用数据库，无论服务器使用何种编程语言，服务器都可以利用 SQL 使用一致的接口访问数据。

最后一个示例程序通过向Chinook示例 SQLite 数据库提供 Web 应用程序和用户界面来演示这一点。Peter Stark 慷慨地维护 Chinook 数据库，作为SQLite 教程站点的一部分。如果您想了解有关 SQLite 和 SQL 的更多信息，那么该站点是一个很好的资源。

Chinook 数据库按照简化的Spotify提供艺术家、音乐和播放列表信息。数据库是project/data文件夹中示例代码项目的一部分。

在 Python、SQLite 和 SQLAlchemy 中使用 Flask
该examples/example_3/chinook_server.py程序创建了一个Flask应用程序，您可以使用浏览器与之交互。该应用程序使用以下技术：

Flask Blueprint是Flask的一部分，它提供了一种遵循关注点分离设计原则并创建不同模块以包含功能的好方法。

Flask SQLAlchemy是Flask的扩展，它在您的 Web 应用程序中添加了对 SQLAlchemy 的支持。

Flask_Bootstrap4打包了Bootstrap前端工具包，将其与您的 Flask Web 应用程序集成。

Flask_WTF扩展烧瓶WTForms，让您的网络应用程序来生成和验证Web表单的有效途径。

python_dotenv是一个 Python 模块，应用程序使用它从文件中读取环境变量并将敏感信息保留在程序代码之外。

尽管对于此示例不是必需的，但.env文件包含应用程序的环境变量。该.env文件包含敏感信息，如密码，您应该将其保留在代码文件之外。但是，项目.env文件的内容如下所示，因为它不包含任何敏感数据：

SECRET_KEY = "you-will-never-guess"
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLAlCHEMY_ECHO = False
DEBUG = True

示例应用程序相当大，只有其中的一部分与本教程相关。出于这个原因，检查和学习代码留给读者作为练习。也就是说，您可以查看下面应用程序的动画屏幕截图，然后是呈现主页的 HTML 和提供动态数据的 Python Flask 路由。

这是运行中的应用程序，浏览各种菜单和功能：

动画屏幕截图从应用程序主页开始，使用Bootstrap 4设计样式。该页面显示数据库中的艺术家，按升序排序。屏幕截图的其余部分显示单击显示的链接或从顶级菜单导航应用程序的结果。

这是生成应用程序主页的Jinja2 HTML 模板：

{% extends "base.html" %}

{% block content %}
<div class="container-fluid">
  <div class="m-4">
    <div class="card" style="width: 18rem;">
      <div class="card-header">Create New Artist</div>
      <div class="card-body">
        <form method="POST" action="{{url_for('artists_bp.artists')}}">
          {{ form.csrf_token }}
          {{ render_field(form.name, placeholder=form.name.label.text) }}
          <button type="submit" class="btn btn-primary">Create</button>
        </form>
      </div>
    </div>
    <table class="table table-striped table-bordered table-hover table-sm">
      <caption>List of Artists</caption>
      <thead>
        <tr>
          <th>Artist Name</th>
        </tr>
      </thead>
      <tbody>
        {% for artist in artists %}
        <tr>
          <td>
            <a href="{{url_for('albums_bp.albums', artist_id=artist.artist_id)}}">
              {{ artist.name }}
            </a>
          </td>
        </tr>
        {% endfor %}
      </tbody>
    </table>
  </div>
</div>
{% endblock %}

这是 Jinja2 模板代码中发生的事情：

• 第 1 行使用 Jinja2 模板继承从base.html模板构建此模板。该base.html模板包含所有 HTML5 样板代码以及在站点的所有页面上一致的 Bootstrap 导航栏。

• 第 3 到 37 行包含页面的块内容，它被合并到base.html基本模板中的同名 Jinja2 宏中。

• 第 9 到 13 行渲染表单以创建新艺术家。这使用Flask-WTF的特性来生成表单。

• 第 24 到 32 行创建了一个for循环，用于呈现艺术家姓名表。

• 第 27 到 29行将艺术家姓名呈现为指向艺术家专辑页面的链接，显示与特定艺术家相关联的歌曲。

这是呈现页面的 Python 路由：

from flask import Blueprint, render_template, redirect, url_for
from flask_wtf import FlaskForm
from wtforms import StringField
from wtforms.validators import InputRequired, ValidationError
from app import db
from app.models import Artist

# Set up the blueprint
artists_bp = Blueprint(
    "artists_bp", __name__, template_folder="templates", static_folder="static"
)

def does_artist_exist(form, field):
    artist = (
        db.session.query(Artist)
        .filter(Artist.name == field.data)
        .one_or_none()
    )
    if artist is not None:
        raise ValidationError("Artist already exists", field.data)

class CreateArtistForm(FlaskForm):
    name = StringField(
        label="Artist's Name", validators=[InputRequired(), does_artist_exist]
    )

@artists_bp.route("/")
@artists_bp.route("/artists", methods=["GET", "POST"])
def artists():
    form = CreateArtistForm()

    # Is the form valid?
    if form.validate_on_submit():
        # Create new artist
        artist = Artist(name=form.name.data)
        db.session.add(artist)
        db.session.commit()
        return redirect(url_for("artists_bp.artists"))

    artists = db.session.query(Artist).order_by(Artist.name).all()
    return render_template("artists.html", artists=artists, form=form,)

让我们回顾一下上面的代码是做什么的：

• 第 1 到 6 行导入呈现页面所需的所有模块，并使用数据库中的数据初始化表单。

• 第 9 到 11 行创建了艺术家页面的蓝图。

• 第 13 到 20 行为Flask-WTF 表单创建了一个自定义验证器函数，以确保创建新艺术家的请求不会与现有艺术家发生冲突。

• 第 22 到 25 行创建表单类来处理在浏览器中呈现的艺术家表单并提供表单字段输入的验证。

• 第 27 到 28行将两条路由连接到artists()它们所装饰的函数。

• 第 30 行创建了一个CreateArtistForm()类的实例。

• 第 33 行确定页面是通过 HTTP 方法 GET 还是 POST（提交）请求的。如果它是 POST，那么它还会验证表单的字段并通知用户这些字段是否无效。

• 第 35 到 37 行创建了一个新的艺术家对象，将其添加到 SQLAlchemy 会话中，并将艺术家对象提交到数据库中，并将其持久化。

• 第 38 行重定向回艺术家页面，该页面将使用新创建的艺术家重新呈现。

• 第 40行运行 SQLAlchemy 查询以获取数据库中的所有艺术家并按Artist.name.

• 如果 HTTP 请求方法是 GET，则第41 行呈现艺术家页面。

您可以看到大量的功能是由相当少的代码创建的。

创建 REST API 服务器

您还可以创建一个提供REST API的 Web 服务器。这种服务器提供 URL 端点响应数据，通常是JSON格式。JavaScript 单页 Web 应用程序可以通过使用 AJAX HTTP 请求来使用提供 REST API 端点的服务器。

Flask 是创建 REST 应用程序的绝佳工具。有关使用 Flask、Connexion 和 SQLAlchemy 创建 REST 应用程序的多部分系列教程，请查看Python REST APIs With Flask、Connexion 和 SQLAlchemy。

如果您是 Django 的粉丝并且对创建 REST API 感兴趣，请查看Django Rest Framework – 简介和使用 Django Tastypie 创建超级基本 REST API。

为您的 Python 应用程序考虑 SQLite 是非常值得的，无论它是什么。使用数据库可为您的应用程序提供多功能性，并且可能会为添加其他功能创造惊人的机会。

结论

您在本教程中涵盖了很多关于数据库、SQLite、SQL 和 SQLAlchemy 的基础知识！您已经使用这些工具将平面文件中包含的数据移动到 SQLite 数据库，使用 SQL 和 SQLAlchemy 访问数据，并通过 Web 服务器提供该数据。

在本教程中，您学习了：

• 为什么SQLite 数据库可以成为平面文件数据存储的引人注目的替代品
• 如何规范化数据以减少数据冗余并提高数据完整性
• 如何使用SQLAlchemy以面向对象的方式处理数据库
• 如何构建一个Web 应用程序来为多个用户提供数据库服务

使用数据库是处理数据的强大抽象，它为您的 Python 程序增加了重要的功能，并允许您对数据提出有趣的问题。

您可以通过以下链接获取在本教程中看到的所有代码和数据：

进一步阅读

本教程是对使用数据库、SQL 和 SQLAlchemy 的介绍，但关于这些主题还有很多东西需要了解。这些是功能强大、复杂的工具，没有一个教程可以充分涵盖。以下是一些资源，可提供更多信息以扩展您的技能：

• 如果您的应用程序会将数据库暴露给用户，那么避免 SQL 注入攻击是一项重要技能。有关更多信息，请查看使用 Python 防止 SQL 注入攻击。

• 提供对数据库的 Web 访问在基于 Web 的单页应用程序中很常见。要了解如何操作，请查看带有 Flask、Connexion 和 SQLAlchemy 的 Python REST API – 第 2 部分。

• 准备数据工程工作面试会让你在职业生涯中占上风。要开始，请查看Python 的数据工程师面试问题。

• 使用 Flask 与 Postgres 和 SQLAlchemy 迁移数据并能够回滚是软件开发生命周期 (SDLC) 不可或缺的一部分。您可以通过查看Flask by Example – 设置 Postgres、SQLAlchemy 和 Alembic了解更多信息。

【生长吧！Python】有奖征文火热进行中：https://bbs.huaweicloud.cn/blogs/278897