zut.db.maria.MariaDb

class zut.db.maria.MariaDb(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: MysqlDb

__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
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 | None

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: T_Cursor) → int | None

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.