ace.database.sql

Run parameterized SQL queries and return their rows as typed objects.

class ace.database.sql.SqlQuery(name: str, sql: str, result_type: Type[T] | None = None, *, strict: bool = False)

Bases: Generic[T]

Named SQL query whose rows are returned as result_type.

Parameters:

• name – Query name used in errors and logs

• sql – SQL text with ? placeholders

• result_type – Row type (dataclass, pydantic model, dict, tuple, scalar); None returns raw rows

• strict – Reject columns that match no field

__init__(name: str, sql: str, result_type: Type[T] | None = None, *, strict: bool = False)

fetch_all(db_connection: Any, *params: Any) → List[T]

Return all rows.

fetch_one(db_connection: Any, *params: Any) → T | None

Return the first row, or None.

scalar(db_connection: Any, *params: Any) → Any

Return the first column of the first row, or None.

execute(db_connection: Any, *params: Any) → int

Run a statement without result rows and return the row count.

classmethod __class_getitem__(params)

Parameterizes a generic class.

At least, parameterizing a generic class is the main thing this method does. For example, for some generic class Foo, this is called when we do Foo[int] - there, with cls=Foo and params=int.

However, note that this method is also called when defining generic classes in the first place with class Foo(Generic[T]): ….

exception ace.database.sql.SqlMappingError

Bases: ValueError

A row of a query cannot be turned into the query’s result type.

__init__(*args, **kwargs)

add_note()

Exception.add_note(note) – add a note to the exception

args

with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

ace.database.sql.session_user(db_connection: Any, user_id: int | None) → Iterator[None]

Run the block as user_id for row-level security, then restore the previous user.

ace.database.sql.session_service(db_connection: Any)

Run the block as the ACE service user, then restore the previous user.

ace.database.sql.current_session_user(db_connection: Any) → int | None

Return the session UserId used by row-level security, or None.

class ace.database.sql.LegacySqlQuery(db_connection: Any, user_id: int | None = None, as_admin: bool | None = None)

Bases: object

Compatibility class for scripts that defined their own SqlQuery helper.

New code should use SqlQuery and session_user.

__init__(db_connection: Any, user_id: int | None = None, as_admin: bool | None = None)

set_context()

query(sql_query: str, *params: Any)

query_and_fetch(sql_query: str, *params: Any)

query_and_fetch_one(sql_query: str, *params: Any)

query_and_fetch_list_of(sql_query: str, target_type: Type[T], *params: Any) → List[T]

Return all rows, built with the columns as keyword arguments.

query_and_fetch_one_as(sql_query: str, target_type: Type[T], *params: Any) → T

Return the first row; raise IndexError when there is none.

query_and_fetch_as(sql_query: str, target_type: Type[T], *params: Any) → List[T]

Return all rows, setting the columns the type has as attributes.

static rows_to_type(cursor, target_type: Type[T]) → List[T]