如何在数据库类的不同实例之间有效地共享数据库引擎？

Question

序言：不确定这类问题属于stackoverflow还是软件工程，如果我问错地方了，请告诉我。

问题

我正在为 CLI 应用程序开发 DB 包装器。为了使用该包装器（一个类），我只想为每个要映射的表实例化它一次。然而，我目前使用的版本存在问题，每个数据库都会创建多个引擎（每次实例化该类时）。

这是我当前的课程（带有测试的版本位于Code Review）：

from sqlmodel import Session, SQLModel, create_engine, select

class DB:
    def __init__(self, url: str, table: SQLModel, *, echo=False):
        """Database wrapper specific to the supplied database table.

        url: URL of the database file.
        table: Model of the table to which the operations should refer.
                Must be a Subclass of SQLModel.
        """
        self.url = url
        self.table = table
        self.engine = create_engine(url, echo=echo)

    def create_metadata(self):
        """Creates metadata, call only once per database connection."""
        SQLModel.metadata.create_all(self.engine)

    def read_all(self):
        """Returns all rows of the table."""
        with Session(self.engine) as session:
            entries = session.exec(select(self.table)).all()
            return entries

    def read(self, _id):
        """Returns a row of the table."""
        with Session(self.engine) as session:
            entry = session.get(self.table, _id)
            return entry

    def add(self, **fields):
        """Adds a row to the table. Fields must map to the table definition."""
        with Session(self.engine) as session:
            entry = self.table(**fields)
            session.add(entry)
            session.commit()

    def update(self, _id, **updates):
        """Updates a row of the table. Updates must map to the table definition."""
        with Session(self.engine) as session:
            entry = self.read(_id)
            for key, val in updates.items():
                setattr(entry, key, val)
            session.add(entry)
            session.commit()

    def delete(self, _id):
        """Delete a row of the table."""
        with Session(self.engine) as session:
            entry = self.read(_id)
            session.delete(entry)
            session.commit()

使用方式如下：

from db import DB
from models import Project, Account

URL = "sqlite:///database.db"

projects = DB(url=URL, table=Project)
accounts = DB(url=URL, table=Account)

projects.read_all()
accounts.read(4)

但是，

projects

和

accounts

现在使用不同的引擎。我对 ORM 不太熟悉，但我可以想象，这是不可取的。

有没有办法只创建一次引擎并保持类的用法相同，同时仍然能够连接到多个数据库？

解决方案思路（仅当 URL 保持不变时）

我正在考虑这样的事情，但是只有当所有实例的 url 保持相同时才有效。然而，这已经导致我的测试不再通过，因为它们使用每个测试函数都不同的临时数据库。如果我想连接到超过 1 个数据库，这也不再起作用了。

也许还有比这更强大的东西？

class DB:
    engine = None

    def __init__(self, url: str, table: SQLModel, *, echo=False):
        self.url = url
        self.table = table
        if DB.engine is None:
            DB.engine = create_engine(url, echo=echo)

    def create_metadata(self):
        SQLModel.metadata.create_all(self.engine)

可能的替代方案？

我想到的第二个解决方案是这样的，但是它增加了获取数据库连接的额外步骤。不过，它解决了上述解决方案的问题：我只为每个数据库创建一个引擎，同时仍然能够连接到多个数据库。它也适用于我的测试。虽然使用数据库包装器时设置变得复杂，但我现在需要跟踪数据库实例和引擎实例。

class Engine:
    def __init__(self, url, *, echo=False):
        self.url = url
        self.engine = create_engine(url, echo=echo)
    
    def create_metadata(self):
        """Creates metadata, call only once per database connection."""
        SQLModel.metadata.create_all(self.engine)

class DB:
    def __init__(self, table: SQLModel, engine):
        self.table = table
        self.engine = engine.engine

用途：

from db import Engine, DB
from models import Project, Account

URL = "sqlite:///database.db"
engine = Engine(URL)

projects = DB(table=Project, engine=engine)
accounts = DB(table=Account, engine=engine)

projects.read_all()
accounts.read(4)

Answer 1

第二种方法是一个完全有效的解决方案，请参阅依赖注入。

如果您愿意，您可以更进一步，使用类属性将

Engine

的创建“隐藏”在

DB

内。

但这会增加类之间的耦合。

缓解这种情况的一种方法是使

Engine

成为一个适当的单例类。