Python's sqlite3 module provides a solid interface for database operations, but the raw API requires considerable boilerplate code for typical CRUD workflows. This article demonstrates a practical ORM-style wrapper that simplifies database interactions while adding pagination capabilities.

SQLite3 Wrapper with Dictionary Results

The native sqlite3 cursor returns results as tuples, which requires accessing columns by index. A more developer-friendly approach involves converting results to dictionaries, allowing column access by name.

import sqlite3


class SqliteWrapper:
    """SQLite3 database wrapper providing dictionary-style result access."""
    _connection = None

    def __init__(self, database_path: str):
        """Establish connection to the SQLite database.
        
        Args:
            database_path: Path to the SQLite database file.
        """
        self._connection = sqlite3.connect(database_path)

    def _row_to_dict(self, cursor, row):
        """Convert a database row to a dictionary using column names."""
        result = {}
        for index, column in enumerate(cursor.description):
            result[column[0]] = row[index]
        return result

    def execute(self, query: str, parameters: list = None, 
                return_dict: bool = True, commit: bool = True):
        """Execute a SQL query and return results.
        
        Args:
            query: SQL statement with parameter placeholders.
            parameters: List of parameters for the query.
            return_dict: Whether to return results as dictionaries.
            commit: Whether to commit the transaction.
            
        Returns:
            List of rows, either as dictionaries or tuples.
        """
        if return_dict:
            self._connection.row_factory = self._row_to_dict
        else:
            self._connection.row_factory = None

        cursor = self._connection.cursor()
        cursor.execute(query, parameters or [])

        if commit:
            self._connection.commit()

        results = cursor.fetchall()
        cursor.close()
        return results


if __name__ == '__main__':
    db = SqliteWrapper('application.db')
    print(db.execute("SELECT id, username, email FROM users"))
    print(db.execute("SELECT * FROM users", return_dict=False))

Base Repository Class with Reflection

Building on the wrapper, a base class using Python's reflection capabilities can automate common CRUD operations. This approach maps class attributes to database columns, reducing repetitive code across entity classes.

class EntityRepository:
    """Base class providing reflection-based ORM operations."""
    _db_helper = None

    def __init__(self):
        super().__init__()

    @classmethod
    def initialize(cls, database_path: str):
        """Initialize the database connection for all subclasses."""
        EntityRepository._db_helper = SqliteWrapper(database_path)

    def insert(self, entity):
        """Persist an entity instance to its corresponding database table.
        
        The table name must match the class name, and column names 
        must align with instance attributes in declaration order.
        """
        table_name = type(entity).__name__
        query = f'INSERT INTO {table_name} VALUES('
        values = []
        is_first = True

        for attribute in entity.__dict__.keys():
            if is_first:
                is_first = False
                continue
            query += ',?'
            raw_value = getattr(entity, attribute)
            values.append(raw_value if isinstance(raw_value, str) else str(raw_value))

        query = query.replace(',', 'NULL,', 1) + ')'
        EntityRepository._db_helper.execute(query, values)

    def query_by_field(self, entity_class, field_name: str, field_value, 
                       exact_match: bool = True, order_field: str = 'id',
                       use_pagination: bool = False, items_per_page: int = 10,
                       page_number: int = 1):
        """Query entities by a specific field value with optional pagination.
        
        Args:
            entity_class: The entity class to query.
            field_name: Database column name for filtering.
            field_value: Value to match against.
            exact_match: Use exact match (True) or LIKE pattern (False).
            order_field: Column name for ORDER BY clause.
            use_pagination: Enable paginated results.
            items_per_page: Number of items per page when paginating.
            page_number: Which page to retrieve (1-indexed).
            
        Returns:
            List of entity instances matching the criteria.
        """
        table_name = entity_class.__name__
        search_value = field_value if not exact_match else f'%{field_value}%'

        query = f'SELECT * FROM {table_name} WHERE {field_name} = ? ORDER BY {order_field}'
        
        if use_pagination:
            offset = items_per_page * (page_number - 1)
            query += f' LIMIT {items_per_page} OFFSET {offset}'

        results = []
        for row in EntityRepository._db_helper.execute(query, [search_value]):
            instance = entity_class()
            instance.populate(row)
            results.append(instance)

        return results

    def query_first(self, entity_class, field_name: str, field_value, exact_match: bool = True):
        """Query a single entity matching the specified criteria.
        
        Returns:
            First matching entity instance or None if no match found.
        """
        table_name = entity_class.__name__
        search_value = field_value if exact_match else f'%{field_value}%'

        query = f'SELECT * FROM {table_name} WHERE {field_name} = ? LIMIT 1'
        results = EntityRepository._db_helper.execute(query, [search_value])

        if results:
            instance = entity_class()
            instance.populate(results[0])
            return instance
        return None

    def update(self, entity, key_field: str = 'id'):
        """Update an existing entity in the database."""
        table_name = type(entity).__name__
        query = f'UPDATE {table_name} SET '
        parameters = []

        for attribute, value in entity.__dict__.items():
            if attribute == key_field:
                continue
            query += f', {attribute}=?'
            parameters.append(value if isinstance(value, str) else str(value))

        query += f' WHERE {key_field} = ?'
        parameters.append(getattr(entity, key_field))

        EntityRepository._db_helper.execute(query.replace(',', '', 1), parameters)

    def delete(self, entity, key_field: str = 'id'):
        """Remove an entity from the database."""
        table_name = type(entity).__name__
        key_value = getattr(entity, key_field)
        
        query = f'DELETE FROM {table_name} WHERE {key_field}=?'
        EntityRepository._db_helper.execute(query, [str(key_value)])

Entity Base Class and Concrete Implementation

Entities should inherit from a base class that defines the population interface. This contract ensures consistent data mapping across different entity types.

from abc import ABC, abstractmethod
from enum import Enum


class UserLevel(Enum):
    """Enumeration representing user permission levels."""
    STANDARD = "read"
    CONTRIBUTOR = "read,write"
    ADMINISTRATOR = "read,write,delete"


class Entity(ABC):
    """Abstract base class for all database entities."""

    @abstractmethod
    def populate(self, data: dict):
        """Populate instance attributes from a dictionary.
        
        Subclasses must implement this to map database columns
        to their respective attributes.
        """
        pass


class UserAccount(Entity):
    """User account entity mapped to the accounts table."""

    def __init__(self):
        self.id = 0
        self.username = ''
        self.password = ''
        self.display_name = ''
        self.access_level = UserLevel.STANDARD

    def populate(self, data: dict):
        if len(data) == 5:
            self.id = int(data['id'])
            self.username = data['loginname']
            self.password = data['loginpass']
            self.display_name = data['showname']
            self.access_level = UserLevel(data['level'])

Data Access Object Implementation

The DAO pattern provides a clean separation between business logic and database operations. The following implementation demonstrates repository methods for user account operations.

class UserRepository(EntityRepository):
    """Data access object for UserAccount entities."""

    def find_by_id(self, user_id: int):
        """Retrieve a user account by its unique identifier."""
        return self.query_first(UserAccount, 'id', user_id)

    def find_by_username(self, username: str):
        """Retrieve a user account by login username.
        
        Performs exact match by default. Pass exact_match=False
        for pattern matching behavior.
        """
        return self.query_first(UserAccount, 'loginName', username)

    def search_by_level(self, level: str, page: int = 1, items_per_page: int = 20):
        """Search users by access level with pagination support."""
        return self.query_by_field(
            UserAccount, 'level', level,
            exact_match=True, order_field='id',
            use_pagination=True, items_per_page=items_per_page,
            page_number=page
        )

    def create(self, user: UserAccount):
        """Create a new user account in the database."""
        self.insert(user)

    def modify(self, user: UserAccount):
        """Update an existing user account."""
        self.update(user)

    def remove(self, user: UserAccount):
        """Remove a user account from the database."""
        self.delete(user)


if __name__ == '__main__':
    UserRepository.initialize('production.db')
    repository = UserRepository()

    target_user = repository.find_by_username('administrator')
    if target_user:
        print(f"Found user: {target_user.display_name}")

    for user in repository.search_by_level('ADMINISTRATOR', page=1):
        print(f"Admin user: {user.username}")

Database Schema Requirements

The ORM layer expects specific conventions for proper mapping:

• Table names must match entity class names exactly
• Column names must correspond to entity attributes (with potential name mapping)
• The populate() method in each entity handles conversion from data base format
• Primary key should be named id and auto-incrementing

CREATE TABLE accounts (
    id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
    loginname VARCHAR(20) UNIQUE NOT NULL,
    loginpass VARCHAR(100) NOT NULL,
    showname VARCHAR(30) NULL,
    level VARCHAR(100) NOT NULL
);

Extension Considerations

Several enhancements can improve this foundation. Column-to-attribute name mapping allows database columns to differ from entity property names. This requires either decorator-based configuration or a centralized mapping registry. Lazy loading for related entities reduces initial query overhead by deferring relationship resolution until accessed. Transaction management through context managers ensures atomic operations across multiple database changes.

The current implementation prioritizes simplicity and readability. More complex requirements may warrant additional abstraction layers or migration to established ORM frameworks.