DBXS: 为python提供数据库访问层

快速入门

• 从PyPI使用pip install dbxs获取。
• 在dbxs.readthedocs.org上阅读文档。

它是做什么的？为什么我需要它？

DBXS是一个查询组织器。它将您的SQL查询集合放入传统数据访问层，利用Python内置的typing.Protocol提供类型安全和便捷。

DBXS旨在为使用它的应用程序提供5个特性

1. 透明性

DBXS是纯SQL。除了根据所选驱动程序的paramstyle插入占位符之外，它不会解析或解释您提供的SQL语句。它不会生成SQL。您输入的将执行什么。使用您的数据库提供的任何特殊功能，中间没有层。

2. 安全性

使用DBXS难以意外编写SQL注入漏洞。通过要求在导入时间编写SQL语句，在应用程序代码执行之前，攻击者控制的数据不太可能可用。如果您的代码按照DBXS的习惯用法（即：不调用.execute(...)）使用，您可以知道它是安全的。

3. 类型安全

虽然SQL是“原始”的，但您的结果不是。当您使用DBXS声明查询时，您以数据类的形式描述其结果的形状，为您的应用程序代码提供结构化、文档化的数据对象来消费。查询参数和结果都是类型化的。

由于您是生成SQL的人，您仍然负责测试您的查询是否真正消耗和生成了您所声明的类型，但是一旦您编写并测试了一个查询。

4. 结构

当您使用DBXS时，您的所有查询都会收集到定义您数据存储接口的Protocol类中。这为检查数据库查询问题提供了一个小型、清晰的地点集合，既适用于人类读者，也适用于元编程，而不是在整个应用程序中分散数据库关注点。

此外，通过提供简单的异步上下文管理器用于事务管理，通过查找async with transaction而不是查找对commit和rollback的调用，可以更容易地在您的代码中扫描事务边界。

5. 可测试性

dbxs.testing模块提供了对测试数据库接口的最小设置支持。您可以在几行代码中编写一个使用数据库接口的单元测试，如下所示

class MyDBXSTest(TestCase):
    @immediateTest()
    def test_myQuery(self, pool: MemoryPool) -> None:
        async with transaction(pool) as c:
            access = myAccessor(c)
            await access.createFoo("1", "hello")
            self.assertEqual((await access.getFoo("1")).name, "hello")

（由于DBXS仍处于alpha阶段项目，此测试支持目前仅限于SQLite和stdlib unittest，但支持任意数据库驱动和pytest肯定在路线图上。）

示例

使用它看起来像这样

@dataclass
class Quote:
    db: QuoteDB
    id: int
    contents: str

from dbxs import accessor, many, one, query, statement

class QuoteDB(Protocol):
    @query(sql="SELECT id, contents FROM quote WHERE id={id}", load=one(Quote))
    async def quoteByID(self, id: int) -> Quote: ...

    @query(sql="SELECT id, contents FROM quote", load=many(Quote))
    def allQuotes(self) -> AsyncIterable[Quote]: ...

    @statement(sql="INSERT INTO quote (contents) VALUES ({text})")
    async def addQuote(self, text: str) -> None: ...

from dbxs.dbapi import DBAPIConnection
def sqliteWithSchema() -> DBAPIConnection:
    c = connect(":memory:")
    c.execute(
        "CREATE TABLE quote (contents, id INTEGER PRIMARY KEY AUTOINCREMENT)"
    )
    c.commit()
    return c

from dbxs.adapters.dbapi_twisted import adaptSynchronousDriver
driver = adaptSynchronousDriver(sqliteWithSchema, paramstyle)
quotes = accessor(QuoteDB)

async def main() -> None:
    from dbxs.async_dbapi import transaction
    async with transaction(driver) as t:
        quotedb: QuoteDB = quotes(t)
        await quotedb.addQuote("hello, world")
        async for quote in quotedb.allQuotes():
            matched = (await quotedb.quoteByID(quote.id)) == quote
            print(f"quote ({quote.id}) {quote.contents!r} {matched}")

以前的SQL接口解决方案

在Python中与SQL接口时，您有几个选择，这些选择在DBXS之前。

您可以使用低级方法，具体来说是直接使用DB-API 2.0驱动程序。您调用connect(...)以获取连接，然后在那个连接上调用.cursor()以获取游标，最后在该游标上调用.execute，并使用一些SQL和一些参数来告诉数据库执行操作。这种方法的好处是显而易见的；它非常简单直接。如果您想做某事，阅读数据库文档以了解SQL语法，然后输入相应的SQL。

然而，缺点也很明显，主要有两个

1. 没有防止SQL注入的措施：虽然您可以在SQL语句中传递参数，但如果您任何时候遇到一个奇怪的边缘情况，您想在查询中某个参数无法轻易使用的地方放置某些内容（例如，如果您想允许用户选择表名），则没有任何措施可以防止您引入这种漏洞。
2. 每个查询都为您提供了一组扁平的元组列表，“愚蠢”的数据，必须在每个查询点进行解释，而您可能想要某种类型的值对象，它提供方便的方法和命名属性，以及与类型检查器协同工作。

为了减轻这些缺点，您可能使用像Django ORM或SQLAlchemy这样的高级工具。这些工具非常强大，允许您做许多DBXS无法做到的事情，例如动态构建查询。它们提供了强大的便利性，使正确的事情（避免SQL注入）变得容易，并且它们为您提供了强大的高级类型作为查询输入和结果。

然而，这些引入了新的问题

1. 这大大增加了复杂性。为您的项目做出贡献的每个人都需要了解数据库的SQL层以及您的ORM或表达式层。虽然DBXS只包含一些函数和类（主要是query、one、many、statement、transaction和accessor，其中adaptSynchronousDriver是大多数应用程序所需的最多的胶水），一个ORM可能包含数十个甚至数百个额外的函数和数据结构。

2. 这个问题会加剧，因为如果你的数据库有任何你想使用超过“标准”SQL模糊边界的特性，即使是像INSERT…ON CONFLICT DO NOTHING这样简单的操作，你也需要从数据库的SQL方言文档中了解该特性，然后在了解你选择库的特定方言支持时再次学习，假设它支持你想要的特性。有些特性可能只能通过需要依赖的附加库来支持，或者你可能不得不等待它们被实现。

最后，所有这些方法都存在一个共同的缺点：它们都鼓励在应用程序运行时，在方法和函数体中按需定义查询。特别是当使用生成SQL而不是手动指定的高级别库时，这使得从数据库日志中的条目定位到执行它的应用程序代码位置变得非常困难。

类似系统

这听起来熟悉吗？尽管我当时没有听说过它，但我后来了解到，这实际上是JDBI的声明式API的并行发明。如果你注意到这些想法很相似，但术语却完全不同，那就是原因。

限制 & 路线图

据我所知，DBXS（Database eXtension System）大部分还没有在生产环境中使用，因此应被视为alpha级质量。虽然其简单的实现、较小的代码规模和良好的测试覆盖率应该使其生产化变得工作量较小，但它仍然与我认为的“最终”版本相比存在一些主要限制。

• DBXS仅支持异步数据库接口。许多数据库应用程序期望同步接口，[它应该有一个] (https://github.com/glyph/DBXS/issues/18)。

• DBXS仅支持使用Twisted的threadpool包装的同步驱动程序。它应该支持asyncio以及asyncio生态系统中的一些原生数据库驱动程序。