As part of a larger program, I'm trying to find the cleanest way to architecture my program to query a cosmos db and use its return values (dict objects) as easy to manipulate and maintain objects in the whole program.
So I created a parent class called cosmos_objects which inherits from dict and in which I wrote the crud operations and then every class that represents the object stored in each container of my cosmos db inherits from this parent object.
To avoid mistakes while writing the keys from my dictionaries I added them in a subclass Key of my CosmosObjects, every child and grandchild is free to override this subclass to add its own keys
cosmos_db.py :
import os
import logging
from azure.cosmos import CosmosClient, exceptions, PartitionKey, container
from utilities import Singleton
class CosmosDbHandler(metaclass=Singleton):
URI = 'cosmos_db_endpoint_uri'
PRIMARY_KEY = 'cosmos_db_key'
DB = 'ENV'
DEFAULT_PARTITION_KEY = '/id'
def __init__(self):
'''
Initialize the connection with the Cosmos DB and the database
'''
self.endpoint = os.getenv(self.URI)
self.primary_key = os.getenv(self.PRIMARY_KEY)
self.db_name = os.getenv(self.DB)
self.client = CosmosClient(self.endpoint, self.primary_key)
self.database = self.client.get_database_client(database=self.db_name)
def __getattr__(self, _item: str) -> container:
'''
If we try to access an attribute that does not exists, creates the attribute and initialize its value
with the corresponding cosmos db container
'''
self.__dict__[_item] = self.database.get_container_client(_item)
return self.__dict__[_item]
def create_container(self, _container_name: str) -> None:
'''
Creating a new container
'''
try:
self.__dict__[_container_name] = self.database.create_container(id=_container_name,
partition_key=PartitionKey(path=self.DEFAULT_PARTITION_KEY))
except exceptions.CosmosResourceExistsError:
logging.warning(f'Container {_container_name} already exists')
except:
logging.error(f'There was an unknown error creating {_container_name}')
raise
cosmos_object.py :
from __future__ import annotations
import logging
import uuid
from azure.cosmos import exceptions
from cosmos_db import CosmosDbHandler
from utilities import deep_update
class CosmosObject(dict):
CONTAINER_NAME = None
class Key:
ID = 'id'
DELETED = 'deleted'
LAST_UPDATED_BY = 'last_updated_by'
@classmethod
def get_item(cls, _id: str) -> CosmosObject:
try:
container = getattr(CosmosDbHandler(), cls.CONTAINER_NAME)
cosmos_object = container.read_item(item=_id, partition_key=_id)
return cls(cosmos_object)
except:
logging.exception(f'An unknown error occured during the querying of item : {cls.Key.ID} {_id} in {cls.CONTAINER_NAME}')
raise
@classmethod
def query_all(cls) -> CosmosObject:
'''
Get all items in a container
'''
try:
container = getattr(CosmosDbHandler(), cls.CONTAINER_NAME)
query = 'SELECT * FROM c'
items = container.query_items(query=query, enable_cross_partition_query=True)
cosmos_object = [cls(item) for item in items]
except:
logging.exception(f'Unexpected error occured')
return cosmos_object
@classmethod
def query_items(cls,
_key: str = None,
_value: str = None,
_query: str = None,
_params: list[dict[str, str]] = None) -> list[CosmosObject]:
'''
Get items by key or by custom query
Either use _key/_value for a quick SELECT * FROM c WHERE _key=_value
Or _query/_params for custom complex queries
'''
cosmos_objects = []
try:
container = getattr(CosmosDbHandler(), cls.CONTAINER_NAME)
if _key:
_query = f'SELECT * FROM c WHERE c.{_key}=@_value'
_params = [
{'name': '@_value', 'value': _value}
]
items = container.query_items(query=_query, parameters=_params, enable_cross_partition_query=True)
cosmos_objects = [cls(item) for item in items]
except:
logging.exception(f'An unexpected error occured during the request {_query} with parameters {_params}')
return cosmos_objects
def create_item(self) -> CosmosObject:
'''
Creates the current object in base, if it has no id it generates a new one
'''
self[self.Key.DELETED] = False
if not self.get(self.Key.ID):
self[self.Key.ID] = str(uuid.uuid4())
try:
container = getattr(CosmosDbHandler(), self.CONTAINER_NAME)
cosmos_object = container.create_item(body=self)
except exceptions.CosmosResourceExistsError:
logging.warning(f'The item with {self.Key.ID} {self.get(self.Key.ID)} already exists in {self.CONTAINER_NAME}')
cosmos_object = container.read_item(item=self.get(self.Key.ID), partition_key=self.get(self.Key.ID))
except:
logging.exception(f'An unknown error occured during the creation of item : {self.Key.ID} {self.get(self.Key.ID)} in {self.CONTAINER_NAME}')
raise
return self.__class__(cosmos_object)
def replace_item(self) -> CosmosObject:
'''
Merge the current object with the database version of it and persists it
Current object values have precedence over the database version
'''
try:
container = getattr(CosmosDbHandler(), self.CONTAINER_NAME)
db_item = container.read_item(item=self.get(self.Key.ID), partition_key=self.get(self.Key.ID))
merged_item = deep_update(db_item, self)
cosmos_object = container.replace_item(item=merged_item, body=merged_item)
except:
logging.exception(f'An unknown error occured during the replacement of the item: {self.Key.ID} {self.get(self.Key.ID)} in {self.CONTAINER_NAME}')
raise
return self.__class__(cosmos_object)
def upsert_item(self) -> CosmosObject:
'''
Upsert an item, if it has no id, generates it before
'''
self[self.Key.DELETED] = False
if not self.get(self.Key.ID):
self[self.Key.ID] = str(uuid.uuid4())
try:
container = getattr(CosmosDbHandler(), self.CONTAINER_NAME)
db_item = container.read_item(item=self.get(self.Key.ID), partition_key=self.get(self.Key.ID))
except exceptions.CosmosResourceNotFoundError:
logging.info(f'Item {self.Key.ID} {self.get(self.Key.ID)} in {self.CONTAINER_NAME} does not exist, inserting.')
cosmos_object = container.upsert_item(body=self)
except:
logging.exception(f'An unknown error occured during the upsert of the item: {self.Key.ID} {self.get(self.Key.ID)} in {self.CONTAINER_NAME}')
raise
else:
logging.info(f'Item {self.Key.ID} {self.get(self.Key.ID)} in {self.CONTAINER_NAME} already exists, updating.')
merged_item = deep_update(db_item, self)
cosmos_object = container.upsert_item(body=merged_item)
return self.__class__(cosmos_object)
def delete_item(self) -> None:
'''
Delete the current cosmosObject from the Cosmos database
'''
try:
container = getattr(CosmosDbHandler(), self.CONTAINER_NAME)
container.delete_item(item=self.get(self.Key.ID), partition_key=self.get(self.Key.ID))
except exceptions.CosmosResourceNotFoundError:
logging.warning(f'Impossible to delete item {self.CONTAINER_NAME} : {self.get(self.Key.ID)} because it does not exist')
except:
logging.exception(f'Error deleting item {self.CONTAINER_NAME} : {self.get(self.Key.ID)}')
def soft_delete_item(self) -> None:
'''
Soft deletes the current cosmosObject (it just sets a deleted flag to true)
'''
self[self.Key.DELETED] = True
self.replace_item()
customer.py :
from __future__ import annotations
from cosmos_object import CosmosObject
class Customer(CosmosObject):
''' Helper classes for manipulating customers in cosmos db '''
CONTAINER_NAME = 'Customers'
class Key(CosmosObject.Key):
CUSTOMER_NAME = 'customer_name'
PROJECT_ID = 'project_id'
CUSTOMER_TYPE = 'customer_type'
@classmethod
def query_by_name(cls, _name: str) -> Customer:
'''
Get a customer by its name
'''
res = cls.query_items(_key=cls.Key.CUSTOMER_NAME, _value=_name)
return res[0] if res else None
utilities.py :
class Singleton(type):
'''
Singleton pattern
'''
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super(Singleton, cls).__call__(*args,
**kwargs)
return cls._instances[cls]
def deep_update(mapping: dict, *updating_mappings: dict) -> dict:
'''
Allows deep updating of dictionnaries
'''
updated_mapping = mapping.copy()
for updating_mapping in updating_mappings:
for k, v in updating_mapping.items():
if k in updated_mapping and isinstance(updated_mapping[k], dict) and isinstance(v, dict):
updated_mapping[k] = deep_update(updated_mapping[k], v)
else:
updated_mapping[k] = v
return updated_mapping
start.py :
import os
from customer import Customer
os.environ['cosmos_db_endpoint_uri'] = 'https:xxx'
os.environ['cosmos_db_key'] = 'xxx'
os.environ['ENV'] = 'DEV'
# Call a few methods to test the code
customer_dict = {
"customer_name": "Test",
"project_id": 1
}
customer = Customer(customer_dict).create_item()
print(customer)
'''
{
'customer_name': 'Test',
'project_id': 1,
'deleted': False,
'id': '0eff5104-a256-4c06-9963-dd49a37bd5dc'
}
'''
customer = Customer.query_by_name('Test')
print(customer)
'''
{
'customer_name': 'Test',
'project_id': 1,
'deleted': False,
'id': '0eff5104-a256-4c06-9963-dd49a37bd5dc'
}
'''
customer[Customer.Key.CUSTOMER_TYPE] = 'b2c'
customer.replace_item()
customer = Customer.query_by_name('Test')
print(customer)
'''
{
'customer_name': 'Test',
'project_id': 1,
'deleted': False,
'id': '0eff5104-a256-4c06-9963-dd49a37bd5dc',
'customer_type': 'b2c'
}
'''
customer.delete_item()
customer = Customer.query_by_name('Test')
print(customer)
'''
None
'''
It felt not so bad at first but now that the program has grown a lot, it feels quite bad to have to call myCosmosObject.get(MyCosmosObject.Key.ID) to get the ID.
Also for nested objects in my database, calling myCosmosObject.get(MyCosmosObject.Key.Step.get(Step.Key.StepName)) feels actually awfull.
I have looked into dacite to convert dictionaries to objects, but my problem is I will have to convert back the object to dictionary and to compare it to the JSON that is in base when I want to update it for example.
