ahvn.adapter.db module

class ahvn.adapter.db.ORMUKFAdapter(*args, **kwargs)

基类：BaseUKFAdapter

Simplified ORM adapter that only provides access to entity classes and conversion methods.

This adapter creates SQLAlchemy ORM entities dynamically based on the included fields and provides conversion methods between UKF objects and ORM entities.

virtual_fields = ('id', 'expiration_timestamp')

__init__(*args, **kwargs)

Initialize the ORM adapter with specified field inclusion.

参数:

• name -- Name of the adapter instance.

• include -- List of BaseUKF field names to include in the ORM schema. If None, includes all available BaseUKF fields plus virtual fields. The 'id' field is always included automatically.

• exclude -- List of BaseUKF field names to exclude from the ORM schema. If None, excludes no fields.

• *args -- Additional positional arguments.

• **kwargs -- Additional configuration parameters.

返回:

None

main_table_name()

返回类型:

str

dims_table_name(dim_name)

返回类型:

str

参数:

dim_name (str)

table_names()

Get all table names managed by this adapter.

返回类型:

List[str]

from_ukf(ukf)

Convert a BaseUKF object to a ORMUKFMainEntity object.

参数:

ukf (BaseUKF) -- BaseUKF object to convert.

返回类型:

Dict[str, ExportableEntity]

返回:

A dictionary mapping table names to their corresponding ORM entity instances. main key corresponds to the main entity, and other keys correspond to dimension entities. The main will always be the first item in the dictionary.

entities(kls)

返回类型:

list[BaseUKF]

参数:

kls (Iterable[BaseUKF])

entity_mappings(kls)

Convert a list of BaseUKF objects to dictionary mappings for bulk insert.

This method generates mappings suitable for SQLAlchemy's bulk_insert_mappings(), which is significantly more efficient than add_all() for large batches.

参数:

kls (Iterable[BaseUKF]) -- Iterable of BaseUKF objects to convert.

返回类型:

Dict[str, List[Dict[str, Any]]]

返回:

A dictionary mapping table names to lists of dictionaries. Each dictionary represents a row to be inserted. The 'main' table is always first, followed by dimension tables.

to_ukf_data(entity)

Convert a ORMUKFMainEntity object to a dictionary suitable for BaseUKF initialization.

参数:

entity (ExportableEntity) -- The ORM entity to convert.

返回类型:

Dict[str, Any]

返回:

A dictionary of field names and values for BaseUKF initialization.

from_result(result)

Convert a query result from the backend to the appropriate entity representation.

参数:

result (SQLResponse) -- The raw result from a backend query.

返回类型:

Dict[str, Any]

返回:

The converted entity representation.