Core-Lib provides CRUD ( Create, Read, Update, Delete ) DataAccess classes as boilerplate code.

Example:

from core_lib.data_layers.data_access.db.crud.crud_data_access import CRUDDataAccess
from core_lib.data_layers.data.db.sqlalchemy.base import Base
from core_lib.data_layers.data.db.sqlalchemy.mixins.soft_delete_mixin import SoftDeleteMixin
from core_lib.data_layers.data.db.sqlalchemy.mixins.soft_delete_token_mixin import SoftDeleteTokenMixin
from core_lib.data_transform.result_to_dict import result_to_dict

from sqlalchemy import Column, Integer, VARCHAR, Boolean

class Customer(Base, SoftDeleteMixin, SoftDeleteTokenMixin):
    __tablename__ = 'customer_data'

    id = Column(Integer, primary_key=True, nullable=False)
    name = Column(VARCHAR(length=255), nullable=False, default="")
    email = Column(VARCHAR(length=255), nullable=False, default="")
    active = Column(Boolean, nullable=False)


class CustomerCRUDDataAccess(CRUDDataAccess):

  def __init__(self):
        CRUD.__init__(self, Customer, db_handler, CustomerCRUDDataAccess.rules_validator)

customer = CustomerCRUDDataAccess()

customer.create({'name': 'Jon Doe', 'email': 'abc@def.com', 'active': True})

data = result_to_dict(customer.get(1))
print(data) # {'id': 1, 'name': 'Jon Doe', 'email': 'abc@def.com', 'active': True, 'created_at': 'current_timestamp' , 'updated_at': 'current_timestamp', 'deleted_at': None, 'deleted_at_token': None}

customer.update(1, {'email': 'jon@doe.com'})

data = result_to_dict(customer.get(1))
print(data) # {'id': 1, 'name': 'Jon Doe', 'email': 'jon@doe.com', 'active': True, 'created_at': 'created_timestamp' , 'updated_at': 'current_timestamp', 'deleted_at': None, 'deleted_at_token': None}

customer.delete(1) # will update the updated_at and deleted_at columns in the db with current timestamp and will update the deleted_at_token with timestamp in milliseconds

data = customer.get(1) # will raise StatusCodeException Not found

from core_lib.data_layers.data_access.db.crud.crud import CRUD
from core_lib.data_layers.data_access.db.crud.crud_soft_delete_token_data_access import CRUDSoftDeleteWithTokenDataAccess
from core_lib.data_layers.data.db.sqlalchemy.base import Base
from core_lib.data_layers.data.db.sqlalchemy.mixins.soft_delete_mixin import SoftDeleteMixin
from core_lib.data_layers.data.db.sqlalchemy.mixins.soft_delete_token_mixin import SoftDeleteTokenMixin

from sqlalchemy import Column, Integer, VARCHAR, Boolean

class Customer(Base, SoftDeleteMixin, SoftDeleteTokenMixin):
    __tablename__ = 'customer_data'

    id = Column(Integer, primary_key=True, nullable=False)
    name = Column(VARCHAR(length=255), nullable=False, default="")
    email = Column(VARCHAR(length=255), nullable=False, default="")
    active = Column(Boolean, nullable=False)


class CustomerCRUDSoftDeleteWithTokenDataAccess(CRUDSoftDeleteWithTokenDataAccess):
    def __init__(self):
        CRUD.__init__(self, Customer, db_handler)

Example for CRUD

Implementation with SoftDelete and SoftDeleteToken

CRUD

core_lib.data_layers.data_access.db.crud.CRUD [source]

CRUD is an abstract base class used by CRUDDataAccess.

class CRUD(ABC):

    def __init__(self, db_entity, db: SqlAlchemyConnectionRegistry, rule_validator: RuleValidator = None):
        self._db_entity = db_entity
        self._db = db
        self._rule_validator = rule_validator

Arguments

  • db_entity: The database entity used by this DataAccesss.
  • db (SqlAlchemyConnectionRegistry): Instance of the SqlAlchemyConnectionRegistry used to connect to the database.
  • rule_validator (RuleValidator): Uses the RuleValidator instance to validate data passed into create() and update() methods (optional).

Functions provided by CRUD

get()

core_lib.data_layers.data_access.db.crud.Crud.get() [source]

An abstract method to be implemented with DataAccess subclasses

def get(self, id: int):

Arguments

  • id (int): Id of the column.

delete()

core_lib.data_layers.data_access.db.crud.Crud.delete() [source]

An abstract method to be implemented with DataAccess subclasses

def delete(self, id: int):

Arguments

  • id (int): Id of the column.

create()

core_lib.data_layers.data_access.db.crud.Crud.create() [source]

Is used to add a new entry into the database, this function takes a dict with the data to be added to the database.

def create(self, data: dict):

Arguments

  • data (dict): Key-values pair where key is the column name and value is the entry to be added to the column.

update()

core_lib.data_layers.data_access.db.crud.Crud.update() [source]

Used to update data in the database. This function takes a dict that contains the data to be updated as well as the id of the column that needs to be updated.

def update(self, id: int, data: dict):

Arguments

  • id (int): Takes the id of the column to be updated.
  • data (dict): Key-values pair where key is the column name and value is the entry to be updated.

CRUDDataAccess

core_lib.data_layers.data_access.db.crud.CRUDDataAccess [source]

Extends the class CRUD and implements the get() and delete() methods

class CRUDDataAccess(DataAccess, CRUD):

    def __init__(self, db_entity, db: SqlAlchemyConnectionRegistry, rule_validator: RuleValidator):
        CRUD.__init__(self, db_entity, db, rule_validator)

Functions provided by CRUDDataAccess

get()

core_lib.data_layers.data_access.db.crud.CRUDDataAccess.get() [source]

Overrides the function in CRUD class, used to get data from database for a given id.

def get(self, id: int):

Arguments

  • id (int): The id of the column we want to query.

delete()

core_lib.data_layers.data_access.db.crud.CRUDDataAccess.delete() [source]

Overrides the function in CRUD class, deletes the data for the given id.

def delete(self, id: int):

Arguments

  • id (int): The id of the column to be deleted.

Usage

from core_lib.data_layers.data_access.db.crud.crud import CRUD
from core_lib.data_layers.data_access.db.crud.crud_data_access import CRUDDataAccess
from core_lib.data_layers.data.db.sqlalchemy.base import Base

from sqlalchemy import Column, Integer, VARCHAR, Boolean

class Customer(Base):
    __tablename__ = 'customer_data'

    id = Column(Integer, primary_key=True, nullable=False)
    name = Column(VARCHAR(length=255), nullable=False, default="")
    email = Column(VARCHAR(length=255), nullable=False, default="")
    active = Column(Boolean, nullable=False)


class CustomerCRUDDataAccess(CRUDDataAccess):
    def __init__(self):
        CRUD.__init__(self, Customer, db_handler)

CRUDSoftDeleteDataAccess

core_lib.data_layers.data_access.db.crud.CRUDSoftDeleteDataAccess [source]

Similar to CRUDDataAccess but is used to access and handle soft delete. For this to work the database object class must extend SoftDeleteMixin to create the required columns for soft delete.

class CRUDSoftDeleteDataAccess(DataAccess, CRUD):

    def __init__(self, db_entity, db: SqlAlchemyConnectionRegistry, rule_validator: RuleValidator):
        CRUD.__init__(self, db_entity, db, rule_validator)

Functions provided by CRUDSoftDeleteDataAccess

get()

core_lib.data_layers.data_access.db.crud.CRUDSoftDeleteDataAccess.get() [source]

Overrides the function in CRUD class, used to get data from database for a given id where deleted_at is None.

def get(self, id: int):

Arguments

  • id (int): The id of the column we want to query.

delete()

core_lib.data_layers.data_access.db.crud.CRUDSoftDeleteDataAccess.delete() [source]

Overrides the function in CRUD class, soft deletes the data for the given id, SoftDeleteMixin will set the deleted_at to current timestamp.

def delete(self, id: int):

Arguments

  • id (int): The id of the column to be deleted.

Usage

from core_lib.data_layers.data_access.db.crud.crud import CRUD
from core_lib.data_layers.data_access.db.crud.crud_soft_data_access import CRUDSoftDeleteDataAccess
from core_lib.data_layers.data.db.sqlalchemy.base import Base
from core_lib.data_layers.data.db.sqlalchemy.mixins.soft_delete_mixin import SoftDeleteMixin

from sqlalchemy import Column, Integer, VARCHAR, Boolean

class Customer(Base, SoftDeleteMixin):
    __tablename__ = 'customer_data'

    id = Column(Integer, primary_key=True, nullable=False)
    name = Column(VARCHAR(length=255), nullable=False, default="")
    email = Column(VARCHAR(length=255), nullable=False, default="")
    active = Column(Boolean, nullable=False)


class CustomerCRUDSoftDeleteDataAccess(CRUDSoftDeleteDataAccess):

    def __init__(self):
        CRUD.__init__(self, Customer, db_handler)

CRUDSoftDeleteWithTokenDataAccess

core_lib.data_layers.data_access.db.crud.CRUDSoftDeleteWithTokenDataAccess [source]

This class is similar to CRUDSoftDeleteDataAccess but is uses SoftDeleteMixin as well as SoftDeleteTokenMixin which creates the delete_token column, because indexing a DateTime column is slow, adding the delete_token allows us to index the columns that have been deleted.

class CRUDSoftDeleteWithTokenDataAccess(DataAccess, CRUD):

    def __init__(self, db_entity, db: SqlAlchemyConnectionRegistry, rule_validator: RuleValidator):
        CRUD.__init__(self, db_entity, db, rule_validator)

Functions provided by CRUDSoftDeleteWithTokenDataAccess

get()

core_lib.data_layers.data_access.db.crud.CRUDSoftDeleteWithTokenDataAccess.get() [source]

Overrides the function in CRUD class, used to get data from database for a given id where deleted_at is None.

def get(self, id: int):

Arguments

  • id (int): The id of the column we want to query.

delete()

core_lib.data_layers.data_access.db.crud.CRUDSoftDeleteWithTokenDataAccess.delete() [source]

Overrides the function in CRUD class, soft deletes the data for the given id, SoftDeleteMixin will set the deleted_at to current timestamp and SoftDeleteTokenMixin will set the delete_token to int value of current timestamp.

def delete(self, id: int):

Arguments

  • id (int): The id of the column to be soft deleted and also will update the delete_token with the current milliseconds.