Que：Sneks的SQL 🐍

Que允许您动态生成SQL查询，无需ORM的全部开销。

动机

Que应一个需要动态生成ASGI网络服务的SQL的需求而诞生。我希望能够像使用SQLAlchemy这样的ORM进行动态查询，但又希望有全异步数据库客户端的性能。Que试图填补这个空白。选择您喜欢的连接客户端，让Que来处理SQL。

这是什么？

Que旨在解决一个单一目的：使用纯Python生成符合SQL规范的查询。Que完全没有硬依赖，也不会强制使用特定的数据库客户端或方言。

仍然想为您的连接使用SQLAlchemy？那就去做吧。想使用PyMySQL或psycopg2？Que不会阻止你。想使用aiopg这样的asyncio框架？你品味很高！这个库就是为了你写的。

设计

Que的重点是简洁，只需看看一个简单的SELECT需要什么

>>> import que
>>> select = que.Select(table='foo')
>>> select
Select(table='foo', schema=None, filters=FilterList([]), fields=FieldList([]))
>>> sql, args = select.to_sql()
>>> print(sql)
SELECT
  *
FROM
  foo

Que通过参数化您的SQL并为您格式化参数与您选择的DBAPI客户端一起工作

>>> import que
>>> fields = [que.Field('bar')]
>>> filters = [que.Filter(que.Field('id', 1))]
>>> select = que.Select(table='foo', filters=filters, fields=fields)
>>> sql, args = select.to_sql()
>>> print(sql)
SELECT
  bar
FROM
  foo
WHERE
  id = :1

>>> args
[1]
>>> sql, args = select.to_sql(style=que.NameParamStyle.NAME)
>>> print(sql)
SELECT
  bar
FROM
  foo
WHERE
  id = :id

>>> args
{'id': 1}

Que致力于标准化您的SQL操作API，因此初始化INSERT或UPDATE在功能上与初始化SELECT相同

>>> import que
>>> import dataclasses
>>> import datetime
>>>
>>> @dataclasses.dataclass
... class Foo:
...     bar: str
...     id: int = None
...     created: datetime.datetime = None
... 
>>> new_foo = Foo('blah')
>>> fields = que.data_to_fields(new_foo, exclude=None)
>>> insert = que.Insert(table='foo', fields=fields)
>>> sql, args = insert.to_sql(que.NameParamStyle.NAME)
>>> print(sql)
INSERT INTO
  foo (:colbar)
VALUES
  (:valbar)

>>> args
{'colbar': 'bar', 'valbar': 'blah'}

快速入门

Que没有依赖项，非常轻量（目前只有约30Kb！），仅包含几百行代码。安装就像pip3 install que-py一样简单。

然后你就可以开始使用了！导入que并享受乐趣🤘

示例

一个简单的客户端，用于生成SQL并插入新条目

import dataclasses
import sqlite3

import que

@dataclasses.dataclass
class Spam:
    flavor: str
    id: int = None
    created_on: int = None


class SpamClient:
    """A database client for tracking spam flavors."""

    def __init__(self):
        self.conn = sqlite3.connect('sqlite://spam.db')

    def insert_spam(self, spam: Spam):
        fields = que.data_to_fields(spam, exclude=None)
        insert = que.Insert('spam', fields=fields)
        sql, args = insert.to_sql()
        return self.conn.execute(sql, args)

    def get_spam(self, **kwargs):
        fields = que.data_to_fields(kwargs)
        filters = [que.Filter(x) for x in fields]
        select = que.Select('spam', filters=filters)
        return self.conn.execute(*select.to_sql())

    def update_spam(self, spam: Spam):
        fields = [que.Field('flavor', spam.flavor)]
        filters = [que.Filter(que.Field('id', spam.id))]
        update = que.Update('spam', filters=filters, fields=fields)
        return self.conn.execute(*update.to_sql())

    def delete_spam(self, spam: Spam):
        filters = [que.Filter(que.Field('id', spam.id))]
        delete = que.Delete('spam', filters=filters)
        return self.conn.execute(*delete.to_sql())

文档

完整文档即将推出！

愉快的查询🐍

如何贡献

1. 检查开放的问题或创建一个新问题以开始围绕功能想法或错误进行讨论。
2. 在GitHub上为你的问题创建分支或从GitHub 分支来开始对master分支进行更改。
3. 编写一个测试，以证明错误已修复或功能按预期工作。
4. 发送拉取请求，并持续要求维护者将其合并和发布。 :)