CobraPack
Repositories
6
 Projects Packages
3

repository (1.5.3)

Published 2025-10-08 06:35:25 +00:00 by forgejo-bot in CobraPack/repository

Installation

pip install --index-url  repository

About this package

Repository layer for Python applications

Repository

PyPI version Python 3.10+ License: MIT

A generic and extensible framework for implementing repository patterns in Python with asynchronous support. This library provides abstract base classes and interfaces for handling CRUD (Create, Read, Update, Delete) operations on data models, with support for both SQL databases and REST APIs.

Description

The repository pattern is a design pattern that creates an abstraction layer between the data access logic and the business logic. This library provides a comprehensive implementation that supports:

Core Interfaces and Base Classes

  • IRepository: A generic interface serving as a base for all repository types.
  • BaseGetRepository: Abstract base class for retrieving data. Includes methods for:
    • Fetching all items (get_all)
    • Retrieving data in chunks (iter_get)
    • Fetching single items (get_first and get_one)
  • BaseCreateRepository: Abstract base class for creating new records. Supports:
    • Single creation (create)
    • Bulk creation (bulk_create)
    • Chunked asynchronous processing (iter_bulk_create)
  • BaseUpdateRepository: Abstract base class for updating records with similar functionality to creation:
    • Single updates (update)
    • Bulk updates (bulk_update)
    • Chunked processing (iter_bulk_update)
  • BaseRemoveRepository: Abstract base class for removing records:
    • Single deletions (remove)
    • Bulk deletions (bulk_remove)
    • Chunked processing (iter_bulk_remove)
  • BaseRepository: A comprehensive repository combining all CRUD functionalities by inheriting from the individual base classes.

Pre-built Implementations

  • SQL Implementation: Ready-to-use asynchronous SQL repositories built with SQLAlchemy and SQLModel
  • REST Implementation: Base classes for REST API repositories using httpx and cloudscraper

Features

  • Asynchronous Support: Leverages async/await for efficient I/O operations with both database and HTTP requests
  • Generic Design: Uses Python generics to allow type safety and flexibility across different data models
  • Chunked Processing: Enables large datasets to be processed in chunks to optimize memory and performance
  • Extensibility: Abstract methods enforce implementation, ensuring custom repositories adhere to the defined patterns
  • Dual Backend Support: Pre-built implementations for both SQL databases and REST APIs
  • Memory Efficient: Uses generators for large dataset processing to minimize memory consumption
  • Flexible Configuration: Supports custom session factories and client configurations

Installation

You can install the package from PyPI:

pip install repository

Or install directly from our custom package index:

pip install --index-url https://forgejo.3dcra.eu/api/packages/CobraPack/pypi/simple/ repository

For specific backend support:

# For SQL support
pip install repository[sql]

# For REST API support
pip install repository[rest]

# For all features
pip install repository[all]

Quick Start

SQL Repository Example

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlmodel import SQLModel, Field
from repository import BaseGetAsyncSqlRepository, BaseCreateAsyncSqlRepository

class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str

class UserAsyncRepository(
    BaseGetAsyncSqlRepository[User],
    BaseCreateAsyncSqlRepository[User]
):
    model = User
    
    def __init__(self, async_session: AsyncSession):
        super().__init__(async_session=async_session)

# Usage
engine = create_async_engine("sqlite+aiosqlite:///example.db")
async_session = AsyncSession(engine)

repo = UserAsyncRepository(async_session)
user = User(name="John Doe", email="john@example.com")
created_user = await repo.create(user)
all_users = await repo.get_all()

REST Repository Example

from pydantic import BaseModel
from repository import BaseRestApiRepository

class User(BaseModel):
    id: int
    name: str
    email: str

class UserRestRepository(BaseRestApiRepository[User]):
    async def get_user(self, user_id: int) -> User:
        response = await self.async_requests_client.get(f"/users/{user_id}")
        return User(**response.json())

# Usage
repo = UserRestRepository()
user = await repo.get_user(1)

Project Structure

repository/
├── base.py                 # Abstract base classes and interfaces
├── sql/                    # SQL implementation
│   ├── async_repo/        # Asynchronous SQL repositories
│   │   ├── base_async_sql_repo.py
│   │   ├── get_sql_repo.py
│   │   ├── create_sql_repo.py
│   │   ├── update_sql_repo.py
│   │   └── remove_sql_repo.py
│   └── utils.py
└── rest/                   # REST API implementation
    ├── base_rest_repo.py
    └── utils.py

Requirements

  • Python 3.10+
  • Dependencies vary based on chosen backend:
    • SQL: sqlmodel, sqlalchemy
    • REST: httpx, cloudscraper, pydantic

License

This project is licensed under the MIT License - see the LICENSE file for details.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Requirements

Requires Python: >=3.10
Details
PyPI
CobraPack/repository
2025-10-08 06:35:25 +00:00
242
70 KiB
Assets (2)
repository-1.5.3-py3-none-any.whl 11 KiB
repository-1.5.3.tar.gz 58 KiB
Versions (3) View all
1.5.3 2025-10-08
1.5.2 2025-10-08
1.5.1 2025-08-31
Issues