Database modules

Mongodb

Database layer

Reference:

MongoEngine User Documentation

sni.db.mongodb.get_pymongo_collection(collection_name: str, client: Optional[pymongo.mongo_client.MongoClient] = None) → pymongo.collection.Collection

Returns a pymongo collection handler.

sni.db.mongodb.init_mongodb()

Connects to the MongoDB instance.

Does not return anything. Any call to mongoengine will act on that connection. It’s magic.

sni.db.mongodb.new_pymongo_client() → pymongo.mongo_client.MongoClient

Connects to the MongoDB database using pymongo and returns a client object.

See also

pymongo.mongo_client.MongoClient documentation

Mongodb signal handlers

Database signals. See Mongoengine signals

sni.db.signals.on_pre_save(_sender: Any, document: mongoengine.document.Document)

If the document has a updated_on, sets it to the current datetime.

Mongodb migration utilities

Database migration utilities

sni.db.migration.ensure_minimum_version(collection: pymongo.collection.Collection, version: int) → None

For all documents in the collection, if its _version field is less than the given version, sets it to that version.

sni.db.migration.finalize_migration(mongoengine_model_class) → None

Finalizes the migration process for a collection.

Example:

collection = start_migration(EsiRefreshToken)
if collection is None:
    return
...
finalize_migration(EsiRefreshToken)

sni.db.migration.has_outdated_documents(collection: pymongo.collection.Collection, schema_version: int) → bool

Tells wether a (pymongo) collection has documents whose _version field have value less than the given schema version.

sni.db.migration.set_if_not_exist(collection: pymongo.collection.Collection, field_name: str, value: Any, *, version: Optional[int] = None) → None

Creates a field with a given value in all documents, if that field does not already exist. If the version kwargs if specified, only update documents of that version.

sni.db.migration.start_migration(mongoengine_model_class) → Optional[pymongo.collection.Collection]

Starts the migration of a collection. If the returned value is None, then the collection is up to date.

Example:

collection = start_migration(EsiRefreshToken)
if collection is None:
    return
...
finalize_migration(EsiRefreshToken)

Redis

Redis related module

sni.db.redis.new_redis_connection() → redis.client.Redis

Returns a new redis connection handler