pynenc_mongo.util.mongo_client

Module Contents

Classes

UpsertOutcome	Outcome of an upsert operation.
UpsertResult	Result of an upsert operation.
RetryableCollection	Proxy for Collection that adds automatic retry to all operations and stores spec.
PynencMongoClient	Singleton MongoDB client for Pynenc, managing connections and collections with retry logic.

Functions

_shorten_index_if_needed	Shorten index name if namespace would exceed MongoDB’s byte limit.
get_conn_key	Generate a unique connection key based on configuration.
get_conn_args	Generate connection arguments for MongoClient based on configuration.
get_mongo_client	Initialize the MongoDB client using configuration.

Data

_MAX_INDEX_NS_BYTES
logger
RETRYABLE_EXCEPTIONS
MONGO_ERROR_INTERRUPTED
MONGO_ERROR_INTERRUPTED_AT_SHUTDOWN
MONGO_ERROR_NOT_PRIMARY_NO_SECONDARY_OK
MONGO_ERROR_NOT_PRIMARY_OR_SECONDARY
RETRYABLE_OPERATION_FAILURE_CODES

API

pynenc_mongo.util.mongo_client._MAX_INDEX_NS_BYTES

127

pynenc_mongo.util.mongo_client.logger

‘getLogger(…)’

pynenc_mongo.util.mongo_client.RETRYABLE_EXCEPTIONS

()

pynenc_mongo.util.mongo_client.MONGO_ERROR_INTERRUPTED

11600

pynenc_mongo.util.mongo_client.MONGO_ERROR_INTERRUPTED_AT_SHUTDOWN

11602

pynenc_mongo.util.mongo_client.MONGO_ERROR_NOT_PRIMARY_NO_SECONDARY_OK

13435

pynenc_mongo.util.mongo_client.MONGO_ERROR_NOT_PRIMARY_OR_SECONDARY

13436

pynenc_mongo.util.mongo_client.RETRYABLE_OPERATION_FAILURE_CODES

()

class pynenc_mongo.util.mongo_client.UpsertOutcome

Bases: enum.StrEnum

Outcome of an upsert operation.

Initialization

Initialize self. See help(type(self)) for accurate signature.

INSERTED

‘auto(…)’

UPDATED

‘auto(…)’

DUPLICATE

‘auto(…)’

class pynenc_mongo.util.mongo_client.UpsertResult

Bases: typing.NamedTuple

Result of an upsert operation.

outcome: pynenc_mongo.util.mongo_client.UpsertOutcome

None

matched_count: int

0

modified_count: int

0

upserted_id: bson.objectid.ObjectId | None

None

property success: bool

Whether the operation succeeded.

property changed: bool

Whether any document was inserted or modified.

class pynenc_mongo.util.mongo_client.RetryableCollection(collection: pymongo.collection.Collection, spec: pynenc_mongo.util.mongo_collections.CollectionSpec, max_retries: int = 3, base_delay: float = 0.1, max_delay: float = 60.0, max_time: float = 300.0, retry_indefinitely: bool = False)

Proxy for Collection that adds automatic retry to all operations and stores spec.

Initialization

property spec: pynenc_mongo.util.mongo_collections.CollectionSpec

Return the stored CollectionSpec.

__getattr__(name: str) → Any

Proxy all collection methods with retry logic.

_should_stop_retrying(attempt: int, elapsed: float) → bool

Check if retry loop should stop based on attempts and time.

_log_retry(func_name: str, attempt: int, delay: float, error: Exception) → None

Log a warning message for retry attempts.

_log_failure(func_name: str, attempt: int, elapsed: float, error: Exception) → None

Log an error message when retries are exhausted.

_wrap_with_retry(func: collections.abc.Callable[..., Any]) → collections.abc.Callable[..., Any]

upsert_document(filter: dict[str, Any], update: dict[str, Any]) → pynenc_mongo.util.mongo_client.UpsertResult

Upsert a document in the collection.

Parameters:

• filter – The filter to match the document

• update – The document fields to update (will be set)

Returns:

UpsertResult with operation details

insert_or_ignore(document: dict[str, Any]) → None

Insert a document if it does not already exist based on the unique index.

Parameters:

document – The document to insert

class pynenc_mongo.util.mongo_client.PynencMongoClient(conf: pynenc_mongo.conf.config_mongo.ConfigMongo)

Singleton MongoDB client for Pynenc, managing connections and collections with retry logic.

Initialization

_instances: dict[str, pynenc_mongo.util.mongo_client.PynencMongoClient]

None

_lock

‘RLock(…)’

classmethod get_instance(conf: pynenc_mongo.conf.config_mongo.ConfigMongo) → pynenc_mongo.util.mongo_client.PynencMongoClient

Get or create a singleton instance for the given configuration.

property db: pymongo.database.Database

Return the configured database.

list_collection_names() → list[str]

List all collection names in the configured database.

get_collection(spec: pynenc_mongo.util.mongo_collections.CollectionSpec) → pynenc_mongo.util.mongo_client.RetryableCollection

Returns a retryable collection proxy with stored spec.

pynenc_mongo.util.mongo_client._shorten_index_if_needed(index: pymongo.IndexModel, db_name: str, collection_name: str) → pymongo.IndexModel

Shorten index name if namespace would exceed MongoDB’s byte limit.

MongoDB 3.6 limits fully qualified index namespaces (db.collection.$index_name) to 127 bytes. When the auto-generated name would exceed this, a deterministic hash-based name is substituted.

Parameters:

• index – The original IndexModel

• db_name – The database name

• collection_name – The collection name

Returns:

The original index or a new IndexModel with a shortened name

pynenc_mongo.util.mongo_client.get_conn_key(conf: pynenc_mongo.conf.config_mongo.ConfigMongo) → str

Generate a unique connection key based on configuration.

pynenc_mongo.util.mongo_client.get_conn_args(conf: pynenc_mongo.conf.config_mongo.ConfigMongo) → dict[str, str | int | None]

Generate connection arguments for MongoClient based on configuration.

pynenc_mongo.util.mongo_client.get_mongo_client(conf: pynenc_mongo.conf.config_mongo.ConfigMongo) → pymongo.MongoClient

Initialize the MongoDB client using configuration.

Uses mongo_url if defined, otherwise falls back to host/port/username/password/authSource.