ahvn.utils.mdb package

ahvn.utils.mdb.resolve_mdb_config(database=None, collection=None, **kwargs)

Compile a MongoDB configuration dictionary based on the following order of priority: 1. kwargs 2. global configuration When a parameter is specified in multiple places, the one with the highest priority is used.

Parameters:

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

• collection (str, optional) – The collection name to use.

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

Returns:

The resolved MongoDB configuration dictionary.

Return type:

Dict[str, Any]

Example

>>> config = resolve_mdb_config(collection="my_collection")
>>> config = resolve_mdb_config(host="192.168.1.100")
>>> config = resolve_mdb_config(connection_string="mongodb://localhost:27017/mydb")

class ahvn.utils.mdb.BaseMongoType(**kwargs)

Bases: object

Base class for MongoDB field types with UKF conversion.

__init__(**kwargs)

from_ukf(ukf_value)

Convert UKF value to MongoDB value.

Return type:

Any

Parameters:

ukf_value (Any)

to_ukf(mongo_value)

Convert MongoDB value to UKF value.

Return type:

Any

Parameters:

mongo_value (Any)

class ahvn.utils.mdb.MongoIdType(**kwargs)

Bases: BaseMongoType

ID type for MongoDB (_id field).

UKF IDs can be very large integers (beyond 64-bit). MongoDB only supports up to 64-bit integers (8 bytes). We store IDs as strings to avoid overflow.

from_ukf(ukf_value)

Convert UKF ID to MongoDB _id (string).

Return type:

Optional[str]

Parameters:

ukf_value (Any)

to_ukf(mongo_value)

Convert MongoDB _id to UKF ID (integer).

Return type:

Optional[int]

Parameters:

mongo_value (Any)

class ahvn.utils.mdb.MongoTextType(length=None, **kwargs)

Bases: BaseMongoType

Text type for MongoDB (string).

Parameters:

length (int | None)

__init__(length=None, **kwargs)

Parameters:

length (int | None)

from_ukf(ukf_value)

Convert UKF text to MongoDB string.

Return type:

Optional[str]

Parameters:

ukf_value (Any)

to_ukf(mongo_value)

Convert MongoDB string to UKF text.

Return type:

Optional[str]

Parameters:

mongo_value (Any)

class ahvn.utils.mdb.MongoIntegerType(**kwargs)

Bases: BaseMongoType

Integer type for MongoDB.

from_ukf(ukf_value)

Convert UKF integer to MongoDB integer.

Return type:

Optional[int]

Parameters:

ukf_value (Any)

to_ukf(mongo_value)

Convert MongoDB integer to UKF integer.

Return type:

Optional[int]

Parameters:

mongo_value (Any)

class ahvn.utils.mdb.MongoBooleanType(**kwargs)

Bases: BaseMongoType

Boolean type for MongoDB.

from_ukf(ukf_value)

Convert UKF boolean to MongoDB boolean.

Return type:

Optional[bool]

Parameters:

ukf_value (Any)

to_ukf(mongo_value)

Convert MongoDB boolean to UKF boolean.

Return type:

Optional[bool]

Parameters:

mongo_value (Any)

class ahvn.utils.mdb.MongoDurationType(**kwargs)

Bases: BaseMongoType

Duration type for MongoDB (stored as integer seconds).

from_ukf(ukf_value)

Convert UKF timedelta to MongoDB integer (seconds).

Return type:

Optional[int]

Parameters:

ukf_value (timedelta | None)

to_ukf(mongo_value)

Convert MongoDB integer (seconds) to UKF timedelta.

Return type:

Optional[timedelta]

Parameters:

mongo_value (int | None)

class ahvn.utils.mdb.MongoTimestampType(**kwargs)

Bases: BaseMongoType

Timestamp type for MongoDB (stored as integer or datetime).

from_ukf(ukf_value)

Convert UKF datetime to MongoDB integer (Unix timestamp).

Return type:

Optional[int]

Parameters:

ukf_value (datetime | None)

to_ukf(mongo_value)

Convert MongoDB integer (Unix timestamp) to UKF datetime.

Return type:

Optional[datetime]

Parameters:

mongo_value (int | None)

class ahvn.utils.mdb.MongoJsonType(**kwargs)

Bases: BaseMongoType

JSON type for MongoDB (stored as embedded document).

from_ukf(ukf_value)

Convert UKF JSON to MongoDB embedded document.

Return type:

Optional[Dict[str, Any]]

Parameters:

ukf_value (Any)

to_ukf(mongo_value)

Convert MongoDB embedded document to UKF JSON.

Return type:

Any

Parameters:

mongo_value (Dict[str, Any] | None)

class ahvn.utils.mdb.MongoVectorType(**kwargs)

Bases: BaseMongoType

Vector type for MongoDB (stored as array of floats).

from_ukf(ukf_value)

Convert UKF vector to MongoDB array.

Return type:

Optional[List[float]]

Parameters:

ukf_value (List[float] | None)

to_ukf(mongo_value)

Convert MongoDB array to UKF vector.

Return type:

Optional[List[float]]

Parameters:

mongo_value (List[float] | None)

class ahvn.utils.mdb.MongoTagsType(**kwargs)

Bases: BaseMongoType

Tags type for MongoDB (stored as array of {slot, value} subdocuments).

UKF tags are stored as strings like “[slot:value]”. MongoDB stores them as subdocuments: [{“slot”: “…”, “value”: “…”}, …]

from_ukf(ukf_value)

Convert UKF tags (set of “[slot:value]” strings) to MongoDB array.

Return type:

Optional[List[Dict[str, str]]]

Parameters:

ukf_value (set | None)

to_ukf(mongo_value)

Convert MongoDB array to UKF tags (set of “[slot:value]” strings).

Return type:

Optional[set]

Parameters:

mongo_value (List[Dict[str, str]] | None)

class ahvn.utils.mdb.MongoSynonymsType(**kwargs)

Bases: BaseMongoType

Synonyms type for MongoDB (stored as array of strings).

UKF synonyms are already a set of strings.

from_ukf(ukf_value)

Convert UKF synonyms (set of strings) to MongoDB array.

Return type:

Optional[List[str]]

Parameters:

ukf_value (set | None)

to_ukf(mongo_value)

Convert MongoDB array to UKF synonyms (set of strings).

Return type:

Optional[set]

Parameters:

mongo_value (List[str] | None)

class ahvn.utils.mdb.MongoRelatedType(**kwargs)

Bases: BaseMongoType

Related type for MongoDB (stored as array of relation subdocuments).

UKF related are 5-element tuples: (subject_id: int, relation: str, object_id: int, relation_id: Optional[int], relation_resources: Optional[str])

The relation_resources is a JSON string that gets stored in MongoDB as-is.

from_ukf(ukf_value)

Convert UKF related (set of 5-tuples) to MongoDB array.

Return type:

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

Parameters:

ukf_value (set | None)

to_ukf(mongo_value)

Convert MongoDB array of subdocuments to UKF related (set of 5-tuples).

Return type:

Optional[set]

Parameters:

mongo_value (List[Dict[str, Any]] | None)

class ahvn.utils.mdb.MongoAuthsType(**kwargs)

Bases: BaseMongoType

Authorities type for MongoDB (stored as array of subdocuments).

UKF stores auths as set of “[user:authority]” strings. We parse this into subdocuments with user and authority fields.

from_ukf(ukf_value)

Convert UKF auths (set of “[user:authority]” strings) to MongoDB array.

Return type:

Optional[List[Dict[str, str]]]

Parameters:

ukf_value (set | None)

to_ukf(mongo_value)

Convert MongoDB array to UKF auths (set of “[user:authority]” strings).

Return type:

Optional[set]

Parameters:

mongo_value (List[Dict[str, str]] | None)

class ahvn.utils.mdb.MongoCompiler

Bases: object

Compiler that converts KLOp JSON IR to MongoDB MQL.

static compile(expr=None, **kwargs)

Convert a KLOp JSON IR to MongoDB MQL.

Parameters:

• expr (Optional[Dict[str, Any]]) – The parsed filter expression dictionary (optional)

• **kwargs – Filter conditions as key-value pairs

Return type:

Dict[str, Any]

Returns:

MongoDB query expression

Raises:

ValueError – If filter structure is invalid

class ahvn.utils.mdb.MongoDatabase(database=None, collection=None, connect=False, **kwargs)

Bases: object

MongoDB connection wrapper (PyMongo only, sync operations).

Follows the same pattern as Database (utils/db/base.py): - Uses resolve_mdb_config() to get configuration from HEAVEN_CM - Handles connection string building with defaults - Manages connection lifecycle - Provides generic MongoDB access without UKF-specific logic

Example

```python # Use default config mongo = MongoDatabase() collection = mongo.conn

# Override specific params mongo = MongoDatabase(host=”192.168.1.100”, port=27018)

# Use connection string directly mongo = MongoDatabase(connection_string=”mongodb://localhost:27017/mydb”)

# Context manager (auto-close) with MongoDatabase() as mongo:

collection = mongo.mdb[“test”] collection.insert_one({“name”: “Alice”})

```

Note

This class uses PyMongo for synchronous operations only. Motor (async) support will be added in a future phase.

Parameters:

• database (Optional[str])

• collection (Optional[str])

• connect (bool)

__init__(database=None, collection=None, connect=False, **kwargs)

Initialize MongoDB connection.

Similar to Database.__init__(), resolves configuration from HEAVEN_CM.

Parameters:

• database (Optional[str]) – Database name (optional, defaults to config value)

• collection (Optional[str]) – Collection name (optional, defaults to config value)

• connect (bool) – Whether to connect immediately upon initialization.

• **kwargs – Additional connection parameters that override config values - database: Database name (overrides config) - host: MongoDB host (overrides config) - port: MongoDB port (overrides config) - username: Auth username (overrides config) - password: Auth password (overrides config) - connection_string: Full connection string (overrides all) - maxPoolSize, connectTimeoutMS, etc.

connect()

close()

Close MongoDB connection and cleanup resources.

property client: MongoClient

Get PyMongo client for sync operations.

Returns:

The PyMongo client instance.

Return type:

MongoClient

property mdb: Database

Get the specified database collection.

Returns:

The MongoDB database instance.

Return type:

Database

property conn: Collection

Get the specified database collection.

Returns:

The MongoDB collection instance.

Return type:

Collection

create_index(keys, **kwargs)

Create index on collection.

Parameters:

• keys (List[Tuple[str, int]]) – List of (field, direction) tuples direction: 1 for ascending, -1 for descending

• **kwargs – Additional index options (unique, sparse, etc.)

Example

>>> mongo.create_index([("name", 1)], name="users_name_idx")
>>> mongo.create_index([("type", 1), ("age", -1)], name="users_type_age_idx", unique=True)

create_vector_index(embedding_idx, embedding_field, dim)

Create vector search index on collection.

Parameters:

• embedding_idx (str) – Name of the vector index to create.

• embedding_field (str) – Field name containing the vector embeddings.

• dim (int) – Dimensionality of the embedding vectors.

Note

This only works with MongoDB Atlas, not local MongoDB instances.

__enter__()

Context manager entry: returns self.

__exit__(exc_type, exc_val, exc_tb)

Context manager exit: closes connection.

Submodules

• ahvn.utils.mdb.base module
  • MongoDatabase
    • MongoDatabase.__init__()
    • MongoDatabase.connect()
    • MongoDatabase.close()
    • MongoDatabase.client
    • MongoDatabase.mdb
    • MongoDatabase.conn
    • MongoDatabase.create_index()
    • MongoDatabase.create_vector_index()
    • MongoDatabase.__enter__()
    • MongoDatabase.__exit__()
• ahvn.utils.mdb.compiler module
  • MongoCompiler
    • MongoCompiler.compile()
• ahvn.utils.mdb.mdb_utils module
  • resolve_mdb_config()
• ahvn.utils.mdb.types module
  • BaseMongoType
    • BaseMongoType.__init__()
    • BaseMongoType.from_ukf()
    • BaseMongoType.to_ukf()
  • MongoIdType
    • MongoIdType.from_ukf()
    • MongoIdType.to_ukf()
  • MongoTextType
    • MongoTextType.__init__()
    • MongoTextType.from_ukf()
    • MongoTextType.to_ukf()
  • MongoIntegerType
    • MongoIntegerType.from_ukf()
    • MongoIntegerType.to_ukf()
  • MongoBooleanType
    • MongoBooleanType.from_ukf()
    • MongoBooleanType.to_ukf()
  • MongoDurationType
    • MongoDurationType.from_ukf()
    • MongoDurationType.to_ukf()
  • MongoTimestampType
    • MongoTimestampType.from_ukf()
    • MongoTimestampType.to_ukf()
  • MongoJsonType
    • MongoJsonType.from_ukf()
    • MongoJsonType.to_ukf()
  • MongoVectorType
    • MongoVectorType.from_ukf()
    • MongoVectorType.to_ukf()
  • MongoTagsType
    • MongoTagsType.from_ukf()
    • MongoTagsType.to_ukf()
  • MongoSynonymsType
    • MongoSynonymsType.from_ukf()
    • MongoSynonymsType.to_ukf()
  • MongoRelatedType
    • MongoRelatedType.from_ukf()
    • MongoRelatedType.to_ukf()
  • MongoAuthsType
    • MongoAuthsType.from_ukf()
    • MongoAuthsType.to_ukf()