zut.db.pg.PgDb

class zut.db.pg.PgDb(input: T_Connection | str | Mapping | object | None = None, *, name: str | None = None, host: str | None = None, port: int | str | None = None, user: str | None = None, password: str | Secret | None = None, encrypt: bool | None = None, no_autocommit: bool | None = None, tz: tzinfo | str | Literal['local', 'utc'] | None = None, migrations_dir: str | os.PathLike | None = None, commit_migrations: bool | None = None)

Bases: Db[Connection, Cursor]

Database adapter for PostgreSQL (using psycopg (v3) driver).

__init__(input: T_Connection | str | Mapping | object | None = None, *, name: str | None = None, host: str | None = None, port: int | str | None = None, user: str | None = None, password: str | Secret | None = None, encrypt: bool | None = None, no_autocommit: bool | None = None, tz: tzinfo | str | Literal['local', 'utc'] | None = None, migrations_dir: str | os.PathLike | None = None, commit_migrations: bool | None = None)

Methods

__init__([input, name, host, port, user, ...])

add_column(table, columns, *[, ...])

Add column(s) to a table.

alter_column_default(table, columns, *[, ...])

build_connection_url()

Build a connection URL (without password) from a connection object.

check_port()

clear_table(table, *[, truncate, if_exists, ...])

close()

commit()

create_connection(*[, autocommit])

Create a new connection.

create_database(name, *[, if_not_exists, ...])

create_schema(name, *[, if_not_exists, loglevel])

create_table(table, columns, *[, temp, ...])

Create a table with the given columns.

database_exists(name)

drop_column_default(table, columns, *[, ...])

drop_database(name, *[, if_exists, loglevel])

drop_schema(name, *[, if_exists, loglevel])

drop_table(table, *[, if_exists, loglevel])

dump_csv(sql[, params, limit, offset, fmt, ...])

dump_csv_temp(sql[, params, limit, offset, ...])

escape_identifier(value, *[, force])

escape_literal(value)

execute(sql[, params, limit, offset, ...])

execute_file(file[, params, limit, offset, ...])

Execute a SQL script from a file.

execute_file_result(file[, params, limit, ...])

Execute a SQL script from a file and return the results.

execute_function(name[, params, limit, ...])

execute_function_result(name[, params, ...])

execute_procedure(name[, params, limit, ...])

execute_procedure_result(name[, params, ...])

execute_result(sql[, params, limit, offset, ...])

execute_script(script[, params, limit, ...])

execute_script_result(script[, params, ...])

get_column_specs(obj)

get_columns(obj)

get_database_name()

Return the name of the database currently associated with this connection.

get_dict(sql[, params, limit, offset])

get_dicts(sql[, params, limit, offset])

get_foreign_keys(table, *[, columns, ...])

Return the list of foreign keys defined for the given table.

get_last_migration_name()

get_lastrowid(cursor)

The rowid of the last inserted row.

get_namedtuple(sql[, params, limit, offset])

get_namedtuples(sql[, params, limit, offset])

get_now_sql()

get_paginated_and_total_sql(sql, *, limit, ...)

get_paginated_dicts(sql[, params, offset])

Return (rows, total)

get_paginated_namedtuples(sql[, params, offset])

Return (rows, total)

get_paginated_rows(sql[, params, offset])

Return (rows, total)

get_python_type(sql_type)

Get the Python type from a SQL type expressed either as a string, or as an integer (oid in type catalog, for pg and maria)

get_random_table_identifier([prefix, ...])

get_reversed_foreign_keys(columns, table, *)

get_row(sql[, params, limit, offset])

get_rows(sql[, params, limit, offset])

get_scalar(sql[, params, limit, offset, ...])

get_scalars(sql[, params, limit, offset, ...])

get_sql_type(python_type, *[, key, ...])

get_sql_value(value)

Prepare a value so that it can be accepted as input by the database engine.

get_sqlutils_path()

get_unique_keys(table)

get_url(*[, hide_password])

has_sqlutils()

insert_csv(file, table[, columns, create, ...])

Insert data from a source CSV file directly to an existing table, using the most efficient bulk copy method of the database engine without performing any check or conversion.

is_available(*[, migrations])

Try to connect to the database.

is_driver_available()

iter_dicts(sql[, params, limit, offset])

iter_namedtuples(sql[, params, limit, offset])

iter_rows(sql[, params, limit, offset])

iter_scalars(sql[, params, limit, offset, ...])

migrate([migrations_dir, no_sqlutils, ...])

Apply new SQL migration files from the given directory.

parse_identifier(value)

prepare_file_sql(file, *[, encoding, ...])

Read a SQL file and optionally format it with keyword arguments.

prepare_function_sql(name[, params, procedure])

prepare_sql(sql[, params, limit, offset])

print_tabulate(sql, params, ...[, file])

rollback()

schema_exists(name)

single_dict(sql[, params, limit, offset])

single_namedtuple(sql[, params, limit, offset])

single_row(sql[, params, limit, offset])

single_scalar(sql[, params, limit, offset, ...])

table_exists(table, *[, temp])

tabulate(sql[, params, limit, offset])

transaction()

use_database(name)

Attributes

autocommit

bool_sql_type

connection

copy_csv_chunk_size

date_sql_type

datetime_sql_type

decimal_sql_type

default_port

dict_sql_type

driver_availability_error

float_sql_type

in_transaction

int_sql_type

list_sql_type

scheme

sql_type_catalog_by_id

sql_type_catalog_by_name

str_sql_type

strict_types

For sqlite.

uuid_sql_type

vardecimal_sql_type_pattern

varstr_sql_type_pattern

name

host

port

user

password

encrypt

no_autocommit

tz

env_prefix

migrations_dir

commit_migrations

add_column(table: str | tuple[str | None, str] | type[Model], columns: str | Column | Sequence[str | Column], *, ignore_decimal=False, ignore_not_null=False, loglevel=10)

Add column(s) to a table.

NOTE: This method does not intend to manage all cases, but only those usefull for zut library internals.

build_connection_url()

Build a connection URL (without password) from a connection object.

Used by get_url() for external connections (for which URL parts are not given as inputs of the Db object).

create_connection(*, autocommit: bool | None = None)

Create a new connection.

Autocommit:

If not set, used object’s default autocommit.

create_table(table: str | tuple[str | None, str] | type[Model], columns: Iterable[str | Column] | Mapping[str, str | type | Column], *, temp: bool | None = None, if_not_exists=False, primary_key: Iterable[str] | str | bool | None = None, unique_keys: Iterable[str | Sequence[str] | UniqueKey] | None = None, foreign_keys: Sequence[ForeignKey] | dict[str, str] | None = None, ignore_decimal=False, ignore_not_null=False, sql_attributes: str | Sequence[str] | None = None, loglevel=10) → list[Column]

Create a table with the given columns.

Parameters:

sql_attribute – Attributes to be appended at the end of the table. For example, STRICT (for SQLite) or ENGINE=MEMORY (for MySQL/MariaDB).

Returns:

List of columns to create (or that WOULD be created if if_not_exists is True and the table already exists).

execute_file(file: str | os.PathLike, params: Mapping[str, Any] | Sequence[Any] | None = None, *, limit: int | None = None, offset: int | None = None, source: str | None = None, warn_if_result: bool | int | Literal['not-last'] = False, encoding='utf-8', format_kwargs: dict[str, Any] | None = None) → int

Execute a SQL script from a file.

Reads the SQL file, optionally formats it with format_kwargs, and executes it as a script.

Parameters:

  • file – Path to the SQL file to execute.

  • params – Query parameters to bind to the SQL statements.

  • limit – Maximum number of rows to return per statement.

  • offset – Number of rows to skip per statement.

  • source – Source identifier for logging. Defaults to the file’s basename.

  • warn_if_result – Whether to warn if a statement returns results.

  • encoding – File encoding. Defaults to “utf-8”.

  • format_kwargs – Values to substitute into the SQL using Python’s str.format(). WARNING: These values are inserted directly into the SQL string, NOT as parameterized query values. Only use format_kwargs with trusted data to avoid SQL injection attacks.

Returns:

The total number of rows affected by all statements.

execute_file_result(file: str | os.PathLike, params: Mapping[str, Any] | Sequence[Any] | None = None, *, limit: int | None = None, offset: int | None = None, source: str | None = None, warn_if_result: bool | int | Literal['not-last'] = False, encoding='utf-8', format_kwargs: dict[str, Any] | None = None) → ExecuteResult

Execute a SQL script from a file and return the results.

Reads the SQL file, optionally formats it with format_kwargs, and executes it as a script, returning a result manager for accessing query results.

Parameters:

  • file – Path to the SQL file to execute.

  • params – Query parameters to bind to the SQL statements.

  • limit – Maximum number of rows to return per statement.

  • offset – Number of rows to skip per statement.

  • source – Source identifier for logging. Defaults to the file’s basename.

  • warn_if_result – Whether to warn if a statement returns results.

  • encoding – File encoding. Defaults to “utf-8”.

  • format_kwargs – Values to substitute into the SQL using Python’s str.format(). WARNING: These values are inserted directly into the SQL string, NOT as parameterized query values. Only use format_kwargs with trusted data to avoid SQL injection attacks.

Returns:

A ResultManager for accessing the query results.

get_database_name() → str

Return the name of the database currently associated with this connection.

NOTE: - This can be distinct from this class instance attribute name if a statement such as USE has been executed. - This can be None for mysql and maria.

get_foreign_keys(table: str | tuple[str | None, str] | type[Model], *, columns: Iterable[str] | None = None, recurse=False, temp: bool | None = None)

Return the list of foreign keys defined for the given table.

Parameters:

  • table – Source of the foreign key relations.

  • columns – If set, restrict the foreign key searching to the given columns.

  • recurse – If True, check recursively if the related primary keys are themselves part of foreign keys (on the related models).

get_lastrowid(cursor) → int

The rowid of the last inserted row. None if no inserted rowid.

get_paginated_dicts(sql: str, params: Mapping[str, Any] | Sequence[Any] | None = None, *, limit: int, offset: int = 0) → tuple[list[dict[str, Any]], int]

Return (rows, total)

get_paginated_namedtuples(sql: str, params: Mapping[str, Any] | Sequence[Any] | None = None, *, limit: int, offset: int = 0) → tuple[list[NamedTuple], int]

Return (rows, total)

get_paginated_rows(sql: str, params: Mapping[str, Any] | Sequence[Any] | None = None, *, limit: int, offset: int = 0) → tuple[list[tuple], int]

Return (rows, total)

classmethod get_python_type(sql_type: str | int | type | Column | None) → type | None

Get the Python type from a SQL type expressed either as a string, or as an integer (oid in type catalog, for pg and maria)

classmethod get_sql_type(python_type: str | int | type | Column | None, *, key: bool | float = False, ignore_decimal=False) → str
Parameters:

key – indicate whether the column is part of a key (primary or unique). If this is a float, indicate the ratio of the max size of a key to use (for multi column keys).

get_sql_value(value: Any) → Any

Prepare a value so that it can be accepted as input by the database engine.

insert_csv(file: str | os.PathLike | IO[str], table: str | tuple[str | None, str] | type[Model], columns: list[str] | None = None, *, create: bool | Literal['if-not-exists'] = False, primary_key: Iterable[str] | str | bool | None = None, fmt: CsvFormat | str | None = None, encoding: str | None = None, delimiter: str | None = None, loglevel=10) → int

Insert data from a source CSV file directly to an existing table, using the most efficient bulk copy method of the database engine without performing any check or conversion.

Parameters:

  • file – Source CSV file.

  • table – Target table.

  • columns – List of columns to import. Depending on the database engine, this list must or must not be equal to the headers of the CSV file.

  • create – Whether to create the target table.

  • primary_key – Create the target table (if required) with a primary key.

is_available(*, migrations: str | tuple[str, str] | Sequence[str | tuple[str, str]] | Literal['*'] | None = None) → bool

Try to connect to the database. Return True in case of success, False otherwise.

Parameters:

migrations – A list of (app_label, migration_name) indicating Django migration(s) that must be applied. Use ‘*’ if all migrations must be applied.

migrate(migrations_dir: str | PathLike | None = None, *, no_sqlutils=False, commit=False, format_kwargs: dict[str, Any] | None = None)

Apply new SQL migration files from the given directory.

Parameters:

  • migrations_dir – Directory containing the SQL migration files.

  • no_sqlutils – Do not include the common SQL utils files.

  • commit – Commit migrations (applies only if autocommit is disabled).

  • format_kwargs – Values to substitute into the SQL using Python’s str.format(). WARNING: These values are inserted directly into the SQL string, NOT as parameterized query values. Only use format_kwargs with trusted data to avoid SQL injection attacks.

classmethod prepare_file_sql(file: str | PathLike, *, encoding='utf-8', format_kwargs: dict[str, Any] | None = None) → str

Read a SQL file and optionally format it with keyword arguments.

Reads the contents of a SQL file and applies Python’s str.format() with the provided keyword arguments. None values are converted to empty strings.

Parameters:

  • file – Path to the SQL file to read.

  • encoding – File encoding. Defaults to “utf-8”.

  • format_kwargs – Values to substitute into the SQL using Python’s str.format(). WARNING: These values are inserted directly into the SQL string, NOT as parameterized query values. Only use format_kwargs with trusted data to avoid SQL injection attacks. Use the params argument in execute methods for untrusted data.

Returns:

The SQL string with format placeholders substituted.

strict_types = False

For sqlite.