Database Driver Adapters
Database driver adapters in aiosql allow extension of the library to support additional database drivers. If you are using a driver other than the ones currently supported by built-in driver adapters (sqlite3
, apsw
, aiosqlite
, psycopg
, psycopg2
, pg8000
, pygresql
, asyncpg
, pymysql
, mysqlclient
, mysql-connector
) then you will need to make your own. A database driver adapter is a duck-typed class that follows either of the Protocol
types below. These types are defined in aiosql/types.py.
Sync Adapter
class SyncDriverAdapterProtocol(Protocol):
def process_sql(self, query_name: str, op_type: SQLOperationType, sql: str) -> str:
...
def select(
self,
conn: Any,
query_name: str,
sql: str,
parameters: Union[List, Dict],
record_class=Optional[Callable],
) -> List:
...
def select_one(
self,
conn: Any,
query_name: str,
sql: str,
parameters: Union[List, Dict],
record_class=Optional[Callable],
) -> Optional[Any]:
...
def select_cursor(
self, conn: Any, query_name: str, sql: str, parameters: Union[List, Dict]
) -> ContextManager[Any]:
...
def insert_update_delete(
self, conn: Any, query_name: str, sql: str, parameters: Union[List, Dict]
) -> int:
...
def insert_update_delete_many(
self, conn: Any, query_name: str, sql: str, parameters: Union[List, Dict]
) -> int:
...
def insert_returning(
self, conn: Any, query_name: str, sql: str, parameters: Union[List, Dict]
) -> Optional[Any]:
...
def execute_script(self, conn: Any, sql: str) -> None:
...
Async Adapter
class AsyncDriverAdapterProtocol(Protocol):
is_aio_driver = True
def process_sql(self, query_name: str, op_type: SQLOperationType, sql: str) -> str:
...
async def select(
self,
conn: Any,
query_name: str,
sql: str,
parameters: Union[List, Dict],
record_class=Optional[Callable],
) -> List:
...
async def select_one(
self,
conn: Any,
query_name: str,
sql: str,
parameters: Union[List, Dict],
record_class=Optional[Callable],
) -> Optional[Any]:
...
async def select_cursor(
self, conn: Any, query_name: str, sql: str, parameters: Union[List, Dict]
) -> AsyncContextManager[Any]:
...
async def insert_update_delete(
self, conn: Any, query_name: str, sql: str, parameters: Union[List, Dict]
) -> None:
...
async def insert_update_delete_many(
self, conn: Any, query_name: str, sql: str, parameters: Union[List, Dict]
) -> None:
...
async def insert_returning(
self, conn: Any, query_name: str, sql: str, parameters: Union[List, Dict]
) -> Optional[Any]:
...
async def execute_script(self, conn: Any, sql: str) -> None:
...
There isn’t much difference between these two protocols besides the async def
syntax for the method definition. There is one more sneaky difference, the aiosql code expects async adapters to have a static class field is_aio_driver = True
so it can tell when to use await
for method returns. Looking at the source of the builtin adapters/ is a great place to start seeing how you may write your own database driver adapter.
To use the adapter pass its constructor or factory as the driver_adapter
argument when building Queries:
queries = aiosql.from_path("foo.sql", driver_adapter=MyDbAdapter)
Alternatively, an adapter can be registered or overriden:
# in MyDbAdapter provider, eg module "mydb_aiosql"
aiosql.register_adapter("mydb", MyDbAdapter)
# then use it elsewhere
import aiosql
import mydb_aiosql
queries = aiosql.from_path("some.sql", "mydb")
Please ask questions on GitHub Issues. If the community makes additional adapter add-ons I’ll be sure to list them here.