cowbird.database.mongodb

Attributes

StoreSelector

JSON

AnySettingsContainer

SettingsType

MongoDB

MongodbStores

AnyMongodbStore

AnyMongodbStoreType

Classes

DatabaseInterface

Return the unique identifier of db type matching settings.

MonitoringStore

Registry for monitoring instances.

StoreInterface

Check that the store implementation defines its type and index_fields.

MongoDatabase

Return the unique identifier of db type matching settings.

Functions

get_settings(→ cowbird.typedefs.SettingsType)

Retrieve application settings from a supported container.

get_mongodb_connection(→ pymongo.database.Database)

Obtains the basic database connection from settings.

get_mongodb_engine(→ pymongo.database.Database)

Obtains the database with configuration ready for usage.

Module Contents

class cowbird.database.mongodb.DatabaseInterface(_: cowbird.typedefs.AnySettingsContainer)[source]

Return the unique identifier of db type matching settings.

Database interface defining a minimum set of function mostly around store management.

__slots__ = ['type']
static _get_store_type(store_type: cowbird.typedefs.StoreSelector) str[source]
abstract get_store(store_type: cowbird.typedefs.StoreSelector, *store_args: Any, **store_kwargs: Any) cowbird.database.stores.StoreInterface[source]
abstract reset_store(store_type: cowbird.typedefs.StoreSelector) cowbird.database.stores.StoreInterface[source]
abstract get_session() Any[source]
abstract get_information() cowbird.typedefs.JSON[source]
Returns:

{‘version’: version, ‘type’: db_type}

abstract is_ready() bool[source]
cowbird.database.mongodb.StoreSelector[source]
class cowbird.database.mongodb.MonitoringStore(*args: Any, **kwargs: Any)[source]

Bases: StoreInterface, MongodbStore

Registry for monitoring instances.

Uses MongoDB to store what is monitored and by whom.

Init the store used to save monitors.

type = 'monitors'
index_fields = ['callback', 'path']
save_monitor(monitor: cowbird.monitoring.monitor.Monitor) None[source]

Stores Monitor in MongoDB storage.

delete_monitor(monitor: cowbird.monitoring.monitor.Monitor) None[source]

Removes Monitor from MongoDB storage.

list_monitors() List[cowbird.monitoring.monitor.Monitor][source]

Lists all Monitor in MongoDB storage.

clear_services(drop: bool = True) None[source]

Removes all Monitor from MongoDB storage.

class cowbird.database.mongodb.StoreInterface[source]

Bases: object

Check that the store implementation defines its type and index_fields.

type: str = None
index_fields: List[str] = []
cowbird.database.mongodb.JSON[source]
cowbird.database.mongodb.AnySettingsContainer[source]
cowbird.database.mongodb.SettingsType[source]
cowbird.database.mongodb.get_settings(container: cowbird.typedefs.AnySettingsContainer | None, app: bool = False) cowbird.typedefs.SettingsType[source]

Retrieve application settings from a supported container.

Parameters:
  • container – supported container with a handle to application settings.

  • app – allow retrieving from current thread registry if no container was defined.

Returns:

found application settings dictionary.

Raises:

TypeError – when no application settings could be found or unsupported container.

cowbird.database.mongodb.MongoDB: pymongo.database.Database | None = None[source]
cowbird.database.mongodb.MongodbStores[source]
cowbird.database.mongodb.AnyMongodbStore[source]
cowbird.database.mongodb.AnyMongodbStoreType[source]
class cowbird.database.mongodb.MongoDatabase(container: cowbird.typedefs.AnySettingsContainer)[source]

Bases: cowbird.database.base.DatabaseInterface

Return the unique identifier of db type matching settings.

Initialize the mongo database from various type of container.

_database: pymongo.database.Database = None[source]
_settings: cowbird.typedefs.SettingsType = None[source]
_stores: Dict[str, AnyMongodbStore] = None[source]
type = 'mongodb'[source]
reset_store(store_type: AnyMongodbStoreType) AnyMongodbStore | None[source]
get_store(store_type: str | Type[cowbird.database.stores.StoreInterface] | AnyMongodbStoreType, *store_args: Any, **store_kwargs: Any) AnyMongodbStore[source]

Retrieve a store from the database.

Parameters:
  • store_type – type of the store to retrieve/create.

  • store_args – additional arguments to pass down to the store.

  • store_kwargs – additional keyword arguments to pass down to the store.

get_session() Any[source]
get_information() cowbird.typedefs.JSON[source]
Returns:

{‘version’: version, ‘type’: db_type}

is_ready() bool[source]
cowbird.database.mongodb.get_mongodb_connection(container: cowbird.typedefs.AnySettingsContainer) pymongo.database.Database[source]

Obtains the basic database connection from settings.

cowbird.database.mongodb.get_mongodb_engine(container: cowbird.typedefs.AnySettingsContainer) pymongo.database.Database[source]

Obtains the database with configuration ready for usage.