from abc import ABC, abstractmethod from typing import Optional from sqlalchemy.orm import scoped_session from atlas_um.helpers.either import Either from atlas_um.pgdb import pgdb class BaseLogicService(ABC): """ Base interface to encapsulate business logic. Example usage: class ConcreteService(BaseLogicService): def process(self, arg1, arg2): print("Doing some work") ConcreteService.execute(arg1, arg2) """ commit: bool = True @classmethod def execute(cls, *args, **kwargs) -> Optional[Either]: session = pgdb.session instance = cls(session=session) result = instance.process(*args, **kwargs) if cls.commit: session.commit() return result @abstractmethod def process(self, *args, **kwargs) -> Optional[Either]: pass def __init__(self, session: scoped_session): self.session = session def __repr__(self): return self.__class__.__name__