Core-Lib embraces the Onion Architecture 1 2 for code reuse and data flow across libraries. Reusing code and moving logic from one Data-Layer to another is straightforward.

The Data Layers

  1. Data Layer
  2. Data Access Layer
  3. Service Layer

Data Layer

The Data layer defines low-level assets that are used to connect to various third-party sources. For example connection, entities, migration, mapping, etc.

Responsibilities:

  • Define models/entities/connection/migration used by data source

Example define a database entity under the DB connection

/data_layers/data/db/user.py
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


class User(SoftDeleteMixin, Base):
    __tablename__ = 'user'

    id = Column(Integer, primary_key=True, nullable=False)
    email = Column(VARCHAR(length=255), nullable=False)
    ...

SQLAlchemy Structure:

  • This is a common structure from SQLAlchemy for defining a table and mapping table columns
  • Class Definition inheriting from two classes: This is the model class or entity being defined [1].
    • SoftDeleteMixin: Custom class provided by core-lib, offering functionality for soft deletion in SQLAlchemy models.
    • Base: Class facilitating smooth integration of soft deletion functionality into SQLAlchemy models.
    • Parameter order ensures that methods from SoftDeleteMixin take precedence over conflicting methods from the Base class.
  • __tablename__: This specifies the name of the database table associated with this model. In the above case, it’s set to ‘user’.
  • Columns:
    • id: This is a primary key column of type Integer, meaning it holds integer values and uniquely identifies each record in the table.
    • email: This is a column of type VARCHAR, which is a variable-length string type, with a maximum length of 255 characters. It stores the email address of the user.

Data Access Layer

Encapsulate and expose access methods to the Data layer.

Responsibilities:

  • Define APIs to access the Data Layer.
  • Using RuleValidator to validate update and create data.
  • Exception handling happening in the data level.
  • Optimization and speed of data fetching.
class UserDataAccess(DataAccess):

    def __init__(self, db_session: SqlAlchemyConnectionRegistry):
        self.db_session = db_session

    def get(self, id: int):
        with self.db_session.get() as session:
            return session.Query(User).get(id)

Code Explained:

  • Class Definition:
    • UserDataAccess: This is the class being defined.
  • Constructor:
    • __init__(self, db_session: SqlAlchemyConnectionRegistry): This is the constructor method. It initializes instances of UserDataAccess. It takes a parameter db_session of type SqlAlchemyConnectionRegistry, which provides access to the database session.
  • Attributes:
    • self.db_session: This attribute stores the reference to the database session provided during initialization. It’s used to interact with the database.
  • get(self, id: int): This method fetches a user from the database based on the provided id.
    • It first opens a session using self.db_session.get(), which provides a context manager for accessing the database session.
    • Inside the context manager, it queries the database using session.Query(User).get(id). This should fetch a user with the specified id from the User table.

Service Layer

The Service layer provides an API that its users can access. And it will use DataLayer/Other Services/Connection to do so.

Responsibilities:

  • Business Logic

  • Transform return data to dict

  • Caching

from core_lib.data_transform.result_to_dict import ResultToDict
from core_lib.data_layers.service.service import Service

CACHE_KEY_USER = 'key_user_{user_id}'


class UserService(Service):

    def __init__(self, user_data_access: UserDataAccess, user_friends_data_acccess: UserFriendsDataAccess):
        self.user_data_access = user_data_access
        self.user_friends_data_acccess = user_friends_data_acccess

    @Cache(CACHE_KEY_USER)
    @ResultToDict() 
    def get(self, user_id: int):
        user = self.user_data_access.get(user_id)
        if user.something:
            user.friends = self.user_friends_data_acccess.get_user_friends(user_id)
        return user

Code Explained:

  • Constants:
    • CACHE_KEY_USER: This defines a constant string representing the cache key format for user-related data.
  • Class Definition:
    • UserService: This is the class being defined. It provides services related to users.
  • __init__(self, user_data_access: UserDataAccess, user_friends_data_acccess: UserFriendsDataAccess): This is the constructor method. It initializes instances of UserService. It takes two parameters:
    • user_data_access: An instance of UserDataAccess, responsible for accessing user data.
    • user_friends_data_acccess: An instance of UserFriendsDataAccess, responsible for accessing user friends data.
  • Attributes:
    • self.user_data_access: This attribute stores the instance of UserDataAccess.
    • self.user_friends_data_acccess: This attribute stores the instance of UserFriendsDataAccess.
  • Methods:
    • get(self, user_id: int): This method retrieves user data for the given user_id.
    • It decorates the method with @Cache(CACHE_KEY_USER), caching the results based on some cache key format.
    • It also decorates the method with @ResultToDict(), to transform the result into a dictionary format.
    • It retrieves user data using self.user_data_access.get(user_id).
    • It fetches user friends data using self.user_friends_data_acccess.get_user_friends(user_id).
    • Finally, it returns the user data.