This is a fork of sqlc-gen-python v1.3.0 with the following changes:
- Supports
sqlc.embed() - Supports overriding Python types for specific database columns
- Supports SQLAlchemy Session/AsyncSession types
- Enforces timezone-aware datetime types using
pydantic.AwareDatetime - Generates modern Python syntax:
Type | Noneinstead ofOptional[Type]list[T]instead ofList[T]- Adds
_conntype annotations to Querier classes - Imports
IteratorandAsyncIteratorfromcollections.abcinstead oftyping - Assigns unused results to
_variable
- Handles fields with names that conflict with Python reserved keywords
version: "2"
plugins:
- name: py
wasm:
url: https://github.com/asavoy/alt-sqlc-gen-python/releases/download/v0.1.0/alt-sqlc-gen-python.wasm
sha256: TODO
sql:
- schema: "schema.sql"
queries: "query.sql"
engine: postgresql
codegen:
- out: src/authors
plugin: py
options:
package: authors
emit_sync_querier: true
emit_async_querier: trueOptions: emit_sync_querier, emit_async_querier
These options generate Querier and/or AsyncQuerier classes that wrap a SQLAlchemy connection and expose a method for each SQL query.
Querieracceptssqlalchemy.engine.Connection | sqlalchemy.orm.SessionAsyncQuerieracceptssqlalchemy.ext.asyncio.AsyncConnection | sqlalchemy.ext.asyncio.AsyncSession
The query command (:one, :many, :exec, :execrows, :execresult) determines the method signature:
| Command | Sync return type | Async return type |
|---|---|---|
:one |
Model | None |
Model | None |
:many |
Iterator[Model] |
AsyncIterator[Model] |
:exec |
None |
None |
:execrows |
int |
int |
:execresult |
sqlalchemy.engine.Result |
sqlalchemy.engine.Result |
Example generated code with both options enabled:
class Querier[T: sqlalchemy.engine.Connection | sqlalchemy.orm.Session]:
_conn: T
def __init__(self, conn: T):
self._conn = conn
def get_user(self, *, id: int) -> models.User | None:
row = self._conn.execute(sqlalchemy.text(GET_USER), {"p1": id}).first()
if row is None:
return None
return models.User(
id=cast(int, row[0]),
name=cast(str, row[1]),
)
def list_users(self) -> Iterator[models.User]:
result = self._conn.execute(sqlalchemy.text(LIST_USERS))
for row in result:
yield models.User(
id=cast(int, row[0]),
name=cast(str, row[1]),
)
class AsyncQuerier[T: sqlalchemy.ext.asyncio.AsyncConnection | sqlalchemy.ext.asyncio.AsyncSession]:
_conn: T
def __init__(self, conn: T):
self._conn = conn
async def get_user(self, *, id: int) -> models.User | None:
row = (await self._conn.execute(sqlalchemy.text(GET_USER), {"p1": id})).first()
if row is None:
return None
return models.User(
id=cast(int, row[0]),
name=cast(str, row[1]),
)
async def list_users(self) -> AsyncIterator[models.User]:
result = await self._conn.stream(sqlalchemy.text(LIST_USERS))
async for row in result:
yield models.User(
id=cast(int, row[0]),
name=cast(str, row[1]),
)When a query joins multiple tables, you can use sqlc.embed() to nest the full model structs in the result rather than flattening all columns.
Given this schema:
CREATE TABLE authors (
id BIGSERIAL PRIMARY KEY,
name TEXT NOT NULL,
bio TEXT
);
CREATE TABLE books (
id BIGSERIAL PRIMARY KEY,
author_id BIGINT NOT NULL REFERENCES authors(id),
title TEXT NOT NULL,
isbn TEXT
);And this query:
-- name: GetBookWithAuthor :one
SELECT
sqlc.embed(books),
sqlc.embed(authors)
FROM books
JOIN authors ON books.author_id = authors.id
WHERE books.id = $1;The plugin generates a row type with nested model fields:
class GetBookWithAuthorRow(pydantic.BaseModel):
books: models.Book
authors: models.AuthorAnd the querier method constructs each embedded struct from the corresponding columns:
def get_book_with_author(self, *, id: int) -> GetBookWithAuthorRow | None:
row = self._conn.execute(sqlalchemy.text(GET_BOOK_WITH_AUTHOR), {"p1": id}).first()
if row is None:
return None
return GetBookWithAuthorRow(
books=models.Book(
id=cast(int, row[0]),
author_id=cast(int, row[1]),
title=cast(str, row[2]),
isbn=cast(str | None, row[3]),
),
authors=models.Author(
id=cast(int, row[4]),
name=cast(str, row[5]),
bio=cast(str | None, row[6]),
),
)Option: emit_pydantic_models
By default, sqlc-gen-python will emit dataclasses for the models. If you prefer to use pydantic models, you can enable this option.
with emit_pydantic_models
from pydantic import BaseModel
class Author(pydantic.BaseModel):
id: int
name: strwithout emit_pydantic_models
import dataclasses
@dataclasses.dataclass()
class Author:
id: int
name: strOption: emit_str_enum
enum.StrEnum was introduce in Python 3.11.
enum.StrEnum is a subclass of str that is also a subclass of Enum. This allows for the use of Enum values as strings, compared to strings, or compared to other enum.StrEnum types.
This is convenient for type checking and validation, as well as for serialization and deserialization.
By default, sqlc-gen-python will emit (str, enum.Enum) for the enum classes. If you prefer to use enum.StrEnum, you can enable this option.
with emit_str_enum
class Status(enum.StrEnum):
"""Venues can be either open or closed"""
OPEN = "op!en"
CLOSED = "clo@sed"without emit_str_enum (current behavior)
class Status(str, enum.Enum):
"""Venues can be either open or closed"""
OPEN = "op!en"
CLOSED = "clo@sed"Option: overrides
You can override the Python type for specific database columns using the overrides configuration.
column: The fully-qualified column name in the format"table_name.column_name"py_type: The Python type to usepy_import: The module to import the type from
version: "2"
# ...
sql:
- schema: "schema.sql"
queries: "query.sql"
engine: postgresql
codegen:
- out: src/authors
plugin: py
options:
package: authors
overrides:
- column: "authors.id"
py_type: "UUID"
py_import: "uuid"