In SQLAlchemy development, writing separate CRUD operations for each entity class can lead to repetitive code. Drawing inspiration from Java's abstract classes and reflection capabilities, we can implement a generic DAO base class in Python to minimize code duplication. This article presents a straightforward implementation of such a solution.

Entity Class Example

Our entities are simple classes with only attributes and no additional methods. Here's an example:

Base = declarative_base()

class StudentProfile(Base):
    """
    Student profile entity
    Table name: student_profiles
    Stores fundamental student information. During data import, 
    existing records are cleared and replaced with new data.
    """
    __tablename__ = 'student_profiles'
    
    id = Column(Integer, primary_key=True)
    full_name = Column(String(50))
    """Student full name, limited to 50 characters"""
    identification_number = Column(String(18))
    """National ID number, limited to 18 characters"""
    enrollment_year = Column(Integer)
    """Year of enrollment, numeric value"""
    class_number = Column(Integer)
    """Class identifier, numeric value"""
    gender = Column(String(1))
    """Gender, single character"""
    student_id = Column(String(20))
    """Unique student identifier"""

Database Connector

The DatabaseConnector class manages database connections and sessions, using SQLite in this example:

class DatabaseConnector:
    """Database connection factory for managing SQLAlchemy sessions and engines"""
    
    def __init__(self):
        """Initialize with private session and engine variables set to None"""
        self.session = None
        self.engine = None
        self.initialize_database()
    
    def initialize_database(self, db_path="university.db"):
        """
        Configure the database engine
        :param db_path: String specifying the database file location
                       Defaults to 'university.db' in the application directory
        :return: None
        """
        self.engine = create_engine(f'sqlite:///{db_path}', echo=False)
    
    def get_session(self):
        """
        Retrieve a session instance
        Creates a new session if one doesn't exist
        :return: SQLAlchemy Session instance
        """
        if self.engine is None:
            self.initialize_database()
        
        Session = sessionmaker(bind=self.engine)
        if self.session is None:
            self.session = Session()
        return self.session
    
    def create_tables(self):
        """
        Generate database tables based on defined models
        Typically called during application setup
        :return: None
        """
        Base.metadata.create_all(self.engine)

Generic DAO Implementation

The core of our solution is the GenericDAO abstract base class:

from abc import ABC, abstractmethod

class GenericDAO(ABC):
    """
    Abstract base class for Data Access Objects in MVC architecture
    Implements common database operations using SQLAlchemy
    """
    
    def __init__(self):
        """Initialize with a DatabaseConnector instance"""
        self.db_connector = DatabaseConnector()
    
    @abstractmethod
    def get_entity_class(self):
        """
        Abstract method returning the entity class this DAO manages
        Used in query construction
        :return: Entity class
        """
        pass
    
    def find_by_id(self, entity_id):
        """
        Retrieve an entity by its primary key
        :param entity_id: Primary key value of the entity
        :return: Entity instance
        :raises: NoResultError if entity not found
        """
        return self.db_connector.get_session().query(
            self.get_entity_class()
        ).filter(
            self.get_entity_class().id == entity_id
        ).one()
    
    def create(self, entity):
        """
        Persist a new entity to the database
        :param entity: Entity instance to be saved
        :return: None
        """
        self.db_connector.get_session().add(entity)
        self.db_connector.get_session().commit()
    
    def update(self, entity):
        """
        Update an existing entity in the database
        Handles both managed (session-tracked) and detached entities
        
        For managed entities (retrieved within the same session),
        modifications are tracked automatically and only commit is needed.
        
        For detached entities (only ID present), all attributes are compared
        and differences are updated before committing.
        
        :param entity: Modified entity instance
        :raises: EntityNotFoundError if entity doesn't exist
        :raises: EntityTypeMismatch if entity type doesn't match DAO's entity class
        :return: None
        """
        if isinstance(entity, self.get_entity_class()):
            try:
                existing_entity = self.find_by_id(entity.id)
                if existing_entity is None:
                    raise EntityNotFoundError(f"Entity with ID {entity.id} not found")
                
                # Update all non-private attributes that have changed
                for attribute_name in dir(existing_entity):
                    if not attribute_name.startswith('_'):
                        current_value = getattr(entity, attribute_name)
                        existing_value = getattr(existing_entity, attribute_name)
                        if current_value != existing_value:
                            setattr(existing_entity, attribute_name, current_value)
                
                self._commit()
            except Exception as error:
                raise error
        else:
            raise EntityTypeMismatch("Entity type doesn't match DAO's managed entity class")
    
    def _commit(self):
        """Internal method to commit the current session"""
        self.db_connector.get_session().commit()
    
    @abstractmethod
    def delete(self, entity):
        """
        Abstract method for deleting an entity
        :param entity: Entity instance to be deleted
        :return: None
        """
        pass

This implementation provides a clean, reusable foundation for database operations across different entity classes, leveraging Python's abstract base classes and SQLAlchemy's powerful ORM capabilities.