anosql 1.0.1

基础知识

给定一个queries.sql文件：

-- name: get-all-greetings
-- Get all the greetings in the database
SELECT*FROMgreetings;-- name: $select-users
-- Get all the users from the database,
-- and return it as a dict
SELECT*FROMUSERS;

我们可以发出sql查询，例如：

importanosqlimportpsycopg2importsqlite3# PostgreSQLconn=psycopg2.connect('...')queries=anosql.from_path('queries.sql','psycopg2')# Or, Sqlite3...conn=sqlite3.connect('cool.db')queries=anosql.from_path('queries.sql','sqlite3)queries.get_all_greetings(conn)# => [(1, 'Hi')]queries.get_all_greetings.__doc__# => Get all the greetings in the databasequeries.get_all_greetings.sql# => SELECT * FROM greetings;queries.available_queries# => ['get_all_greetings']

参数

通常，您希望动态更改查询的某些部分，特别是 WHERE子句。您可以使用参数来执行此操作：

-- name: get-greetings-for-language-and-length
-- Get all the greetings in the database
SELECT*FROMgreetingsWHERElang=%s;

它们成为位置参数：

visitor_language="en"queries.get_all_greetings(conn,visitor_language)

命名参数

为了使具有许多参数的查询更易于理解和维护，您可以 可以给出参数名称：

-- name: get-greetings-for-language-and-length
-- Get all the greetings in the database
SELECT*FROMgreetingsWHERElang=:langANDlen(greeting)<=:length_limit;

如果您正在编写postgresql查询，还可以将参数格式化为 %s(lang)和%s(length_limit)。

然后，调用查询，就像调用 参数：

visitor_language="en"greetings_for_texting=queries.get_all_greetings(conn,lang=visitor_language,length_limit=140)

更新/插入/删除

为了运行UPDATE、INSERT或DELETE语句，您需要 将!添加到查询名称的末尾。然后anosql将正确执行它。 它还将返回受影响的行数。

插入返回自动生成值的查询

如果希望在运行 插入查询，您可以将<!添加到查询名称的末尾。

-- name: create-user<!
INSERTINTOperson(name)VALUES(:name)

添加自定义查询加载程序。

开箱即用anosql通过stdlibsqlite3数据库驱动程序支持sqlite和postgresql 和psycopg2。如果要扩展anosql以与其他类型的数据库通信 您可以创建一个驱动程序adapeter类并用anosql.register_driver_adapter()注册它。

驱动程序适配器是遵循以下接口的duck类型的类。查看anosql/adapters包 通过查看psycopg2和sqlite3适配器是如何工作的，可以从这里开始。

注册新加载程序：

import anosql

class MyDbAdapter():
    def process_sql(self, name, op_type, sql):
        pass

    def select(self, conn, sql, parameters):
        pass

    @contextmanager
    def select_cursor(self, conn, sql, parameters):
        pass

    def insert_update_delete(self, conn, sql, parameters):
        pass

    def insert_update_delete_many(self, conn, sql, parameters):
        pass

    def insert_returning(self, conn, sql, parameters):
        pass

    def execute_script(self, conn, sql):
        pass


anosql.register_driver_adapter("mydb", MyDbAdapter)

# To use make a connection to your db, and pass "mydb" as the db_type:
import mydbdriver
conn = mydbriver.connect("...")

anosql.load_queries("path/to/sql/", "mydb")
greetings = anosql.get_greetings(conn)

conn.close()

如果适配器构造函数接受参数，则可以注册可以生成 适配器实例：

def adapter_factory():
    return MyDbAdapter("foo", 42)

anosql.register_driver_adapter("mydb", adapter_factory)