ahvn.utils.db package

ahvn.utils.db.resolve_db_config(database=None, provider=None, pool=None, **kwargs)[源代码]

Compile a database configuration dictionary based on the following order of priority: 1. kwargs 2. provider 3. global configuration When a parameter is specified in multiple places, the one with the highest priority is used. When missing, the provider falls back to the default provider.

参数:

  • database (str, optional) -- The database name to use.

  • provider (str, optional) -- The database provider name to use (e.g., 'sqlite', 'pg', 'duckdb').

  • pool (Dict[str, Any], optional) -- Pool configuration to override provider defaults.

  • **kwargs -- Additional parameters to override in the configuration.

返回:

1. The resolved database configuration dictionary with 'url', 'pool', and hyperparameters. Connection parameters (dialect, driver, username, password, host, port, database) are used to build the URL and then removed from the final config. 2. The connection parameters dictionary.

返回类型:

Tuple[Dict[str, Any], Dict[str, Any]]

ahvn.utils.db.create_database_engine(config, conn_args, autocreate=True)[源代码]

Create a SQLAlchemy engine from the resolved database configuration.

Uses appropriate connection pooling strategy based on dialect: - SQLite: StaticPool (file) or SingletonThreadPool (:memory:) - DuckDB: NullPool (thread-safe, no pooling needed) - PostgreSQL/MySQL/MSSQL: QueuePool with configurable settings

Pool settings are read from conn_args['pool'] (set by resolve_db_config from provider config).

参数:

  • config (Dict[str, Any]) -- The database configuration dictionary.

  • conn_args (Optional[Dict[str, Any]]) -- Connection arguments containing dialect, database, pool config, etc.

  • autocreate (bool) -- Whether to automatically create the database if it does not exist.

返回:

A SQLAlchemy engine instance.

返回类型:

Engine

抛出:
ahvn.utils.db.create_database(config, engine_kwargs=None)[源代码]

Create the database if it does not already exist.

This helper supports SQLite (directory creation), PostgreSQL, and MySQL database creation.

参数:

  • config (Dict[str, Any]) -- Database configuration dict containing connection parameters.

  • engine_kwargs (Optional[Dict[str, Any]]) -- Optional kwargs to pass when creating temporary engines.

返回类型:

None

备注

  • The function is best-effort and will log on failure; callers may choose to

    ignore failures by catching exceptions.

ahvn.utils.db.split_sqls(queries, dialect='sqlite')[源代码]

Split a string containing multiple SQL queries into a list.

参数:

  • queries (str) -- The SQL queries to split.

  • dialect (str) -- The SQL dialect to use (default is "sqlite").

返回:

A list of individual SQL queries.

返回类型:

List[str]

ahvn.utils.db.transpile_sql(query, src_dialect='sqlite', tgt_dialect='sqlite')[源代码]

Transpile a SQL query from one dialect to another.

参数:

  • query (str) -- The SQL query to transpile.

  • src_dialect (str) -- The source dialect to transpile from.

  • tgt_dialect (str) -- The target dialect to transpile to (default is the current dialect).

返回:

The transpiled query.

返回类型:

str

抛出:
ahvn.utils.db.load_builtin_sql(query_name, dialect='sqlite', **kwargs)[源代码]

Load SQL query from file and return the query for the current dialect.

警告

This function uses string formatting (.format(**kwargs)) to inject parameters into the SQL query. This is vulnerable to SQL injection if kwargs contains untrusted user input. Only use this function with trusted input or for internal queries where parameters are controlled. For user-supplied values, prefer using parameterized queries supported by your database driver.

参数:

  • query_name (str) -- Name of the SQL file (without .sql extension).

  • dialect (str) -- The SQL dialect to use (default is "sqlite").

  • **kwargs -- Additional parameters for query formatting.

返回:

SQL query for the current dialect. None if the query is not found.

返回类型:

str

抛出:

FileNotFoundError -- If SQL file is not found.

class ahvn.utils.db.SQLProcessor(target_dialect)[源代码]

基类：object

Handles SQL transpilation and parameter normalization across different database dialects.

This class centralizes all SQL processing logic including: - SQL dialect transpilation via SQLGlot - Parameter format normalization (convert all to :param format) - Cross-database parameter binding support

参数:

target_dialect (str)

__init__(target_dialect)[源代码]

Initialize SQL processor for a target database dialect.

参数:

target_dialect (str) -- Target database dialect (sqlite, postgres, mysql, duckdb, etc.)

process_query(query, params=None, transpile_from=None)[源代码]

Process a SQL query and parameters for execution.

参数:

  • query (str) -- Raw SQL query

  • params -- Query parameters (dict, tuple, list, or None)

  • transpile_from (str) -- Source dialect to transpile from (if different from target)

返回类型:

tuple[str, dict]

返回:

Tuple of (processed_query, normalized_params)

Submodules