Felesh Shared Package

A foundational Python package providing scaffold utilities and SDK capabilities for all Felesh microservices. This package ensures consistent architecture patterns, code structure, and development practices across the entire microservices ecosystem.

🎯 Architecture Templates

Choose the architecture that fits your service requirements:

Simple Architecture

• Best for: Traditional CRUD operations, straightforward business logic
• Components: Models, Repositories, GenericServices, UsecaseServices, Views
• Complexity: Low - 50% less boilerplate than CQRS
• Use when: No event publishing, synchronous workflows, fast development needed

Event-Driven CQRS Architecture

• Best for: Event-driven services, distributed transactions, complete audit trails
• Components: Commands, Queries, Handlers, MessageBus, EventStore, Outbox Pattern
• Complexity: Higher - Full CQRS with event sourcing
• Use when: Event publishing required, zero event loss critical, distributed transactions

🚀 Features

Shared Across All Architectures

• Framework Support: Native integration with Django REST Framework and FastAPI
• Unit of Work: Transaction management across repositories
• Repository Pattern: Abstraction for data access
• Type Safety: Full Pydantic V2 integration
• Clean Architecture: Separation of concerns maintained

Event-Driven CQRS Only

• CQRS Pattern: Clean separation between commands (writes) and queries (reads)
• Event-Driven Architecture: Hybrid outbox pattern for guaranteed event delivery with zero loss
• MessageBus: Automatic command/query routing to handlers
• Event Store: Complete audit trail of all domain events

📦 Installation

From Git Repository (Production)

# Install latest from develop branch (HTTPS)
uv add git+https://code.arshai.com/felesh-ai/platform/core-services/shared-package.git

# Install latest from develop branch (SSH)
uv add git+ssh://git@code.arshai.com:2222/felesh-ai/platform/core-services/shared-package.git

# Install specific version/tag
uv add git+https://code.arshai.com/felesh-ai/platform/core-services/shared-package.git@v0.1.0

# Or add to your pyproject.toml

[project.dependencies]
felesh-shared-package = { git = "https://code.arshai.com/felesh-ai/platform/core-services/shared-package.git", branch = "develop" }

From Local Path (Development)

# Install in editable mode for local development
uv add --editable /path/to/felesh_shared_package

# Or add to your pyproject.toml

[project.dependencies]
felesh-shared-package = { path = "../felesh_shared_package", editable = true }

Requirements: Python 3.13+

CLI Commands Available After Installation

Once installed, the package provides the felesh-init command:

# Check if installed correctly
felesh-init --help

# Create a new project from anywhere
felesh-init myservice --framework drf --arch simple

🔧 Development Setup

Code Quality Enforcement

This package uses pre-commit hooks to automatically enforce code quality standards:

# Install pre-commit hooks (one-time setup)
uv run pre-commit install
# or
make precommit-install

Once installed, hooks run automatically on every commit, checking:

• File hygiene (trailing whitespace, EOF newlines, YAML/TOML syntax)
• Ruff linting and formatting (auto-fixes many issues)
• Large file detection and merge conflict markers

Manual execution (optional):

# Run hooks on all files
uv run pre-commit run --all-files
# or
make precommit-run

Makefile targets:

make lint            # Run Ruff linter
make format          # Format code with Ruff
make fix             # Auto-fix lint issues and format
make precommit-run   # Run all pre-commit hooks

🏗️ Project Structure

scaffold/
├── common/                      # Shared CQRS infrastructure
│   ├── seedwork/               # Base classes for CQRS pattern
│   │   ├── commands.py         # Command DTOs and handlers
│   │   ├── queries.py          # Query DTOs and handlers
│   │   ├── events.py           # Event system and dispatching
│   │   ├── uow.py              # Unit of Work pattern
│   │   └── repositories.py     # Repository pattern
│   ├── adapters/               # Framework adapters
│   │   ├── drf/                # Django REST Framework integration
│   │   └── fastapi/            # FastAPI integration
│   └── testing/                # Test utilities and mocks
├── project_templates/          # Ready-to-use project templates
│   ├── drf/                    # DRF project template
│   └── fastapi/                # FastAPI project template
├── examples/                   # Example implementations
│   └── cqrs_patterns/          # CQRS pattern examples
└── init_project.py            # Project initialization script

🚦 Quick Start

Create a New Project

After installing the package, you can use the felesh-init CLI command from anywhere:

# Simple architecture (recommended for CRUD services)
felesh-init myservice --framework drf --arch simple

# Event-Driven CQRS architecture (for event-driven services)
felesh-init myservice --framework fastapi --arch event-driven-cqrs

# Create in specific directory
felesh-init myservice --framework drf --arch simple --path /path/to/projects

# Get help
felesh-init --help

Alternative methods (from felesh_shared_package directory):

# Using Makefile
make init-project NAME=myservice FRAMEWORK=drf ARCH=simple

# Using Python script directly
python scaffold/init_project.py myservice --framework drf --arch simple

Architecture Decision Guide

Criteria	Simple	Event-Driven CQRS
Event Publishing	❌ No cross-service events	✅ Publishes to Kafka
Distributed Transactions	❌ Single database	✅ Eventual consistency
Audit Trail	Standard DB logs	✅ Complete event history
Team Experience	✅ Junior devs OK	⚠️ CQRS knowledge needed
Development Speed	✅ Faster (50% less code)	⚠️ More boilerplate
Use Cases	CRUD, simple workflows	Event-driven, complex sagas

Define Commands and Queries

from scaffold.common.seedwork import ICommand, IQuery
from pydantic import Field, EmailStr

class CreateUserCommand(ICommand):
    """Command to create a new user."""
    email: EmailStr = Field(..., description="User email")
    username: str = Field(..., min_length=3, max_length=50)
    full_name: str = Field(..., description="Full name")

class GetUserByIdQuery(IQuery):
    """Query to get user by ID."""
    user_id: int = Field(..., description="User ID")

Implement Handlers

from scaffold.common.seedwork import (
    ICommandHandler, ICommandResult, Status
)

class CreateUserHandler(ICommandHandler):
    """Handler for CreateUserCommand."""

    def __init__(self, uow, event_collector):
        self.uow = uow
        self.event_collector = event_collector

    def handle(self, command: CreateUserCommand) -> ICommandResult:
        """Execute the create user command."""
        with self.uow:
            # Business logic
            user = self.uow.user_repo.add({
                'email': command.email,
                'username': command.username,
                'full_name': command.full_name,
            })

            # Collect event (saved atomically with data)
            event = UserCreatedEvent(
                user_id=str(user['id']),
                email=user['email']
            )
            self.event_collector.add_event(event)

            self.uow.commit()

            return ICommandResult(
                status=Status.SUCCESS,
                data={'user_id': user['id']}
            )

Use with Django REST Framework

from scaffold.common.adapters.drf import CommandAPIView, QueryAPIView

class CreateUserView(CommandAPIView):
    command_class = CreateUserCommand
    handler_class = CreateUserHandler

class GetUserView(QueryAPIView):
    query_class = GetUserByIdQuery
    handler_class = GetUserByIdHandler

# urls.py
urlpatterns = [
    path('users/', CreateUserView.as_view()),
    path('users/<int:user_id>/', GetUserView.as_view()),
]

Use with FastAPI

from fastapi import FastAPI, Depends
from scaffold.common.adapters.fastapi import (
    create_command_handler_dependency
)

app = FastAPI()

@app.post("/users")
async def create_user(
    command: CreateUserCommand,
    handler = Depends(create_command_handler_dependency(CreateUserHandler))
):
    result = handler.handle(command)
    if result.is_success():
        return result.data
    raise HTTPException(status_code=400, detail=result.message)

📚 Documentation

All documentation is centralized in the /docs directory:

Getting Started

• Index: Complete documentation index with all imports
• Quick Start: Getting started guide
• Architecture: System architecture overview
• Migration: Migration guides

CQRS Pattern

• CQRS Patterns: Commands, Queries, Events DTOs
• Handlers: Base handler classes
• Handler Registry: Handler registration and discovery
• Handler Result: Result types and factory methods

Event System

• Message Bus: Async/Sync message routing
• Event Collector: Event aggregation
• Outbox Pattern: Reliable event publishing
• Inbox Pattern: Exactly-once consumption
• Event System: Event backends and infrastructure

Infrastructure

• Unit of Work: Transaction management
• Repository: Data access patterns
• Models: Base model classes

API Layer

• Filtering: Query filters with operators
• Pagination: Page-based pagination
• API Responses: Response formatting

Framework Adapters

• FastAPI Adapters: FastAPI integration
• DRF Adapters: Django REST Framework integration

Utilities

• Decorators: Handler and service decorators
• Testing: Test strategies and utilities
• Troubleshooting: Common issues and solutions

Examples

• Examples: Working examples of all patterns

🔄 Migration from v1.x

Breaking Changes in v2.0:

• Removed scaffold/drf/seedwork/ and scaffold/fastapi/seedwork/
• All imports now from scaffold.common.seedwork and scaffold.common.adapters
• Service pattern completely removed in favor of CQRS handlers

See Migration Guide for detailed migration instructions.

🎯 Best Practices

1. Separation of Concerns: Keep commands and queries separate
2. Single Responsibility: One handler for one command/query
3. Explicit Transactions: Always use Unit of Work for data modifications
4. Event-Driven: Emit events for important business operations
5. Type Safety: Leverage Pydantic for validation
6. Testing: Mock UoW and repositories, test business logic in isolation

🧪 Testing

def test_create_user_handler():
    # Arrange
    handler = CreateUserHandler()
    handler.uow = Mock(spec=DjangoUnitOfWork)
    handler.uow.user_repo.add.return_value = {'id': 1}

    command = CreateUserCommand(
        email="test@example.com",
        username="testuser",
        full_name="Test User"
    )

    # Act
    result = handler.handle(command)

    # Assert
    assert result.is_success()
    assert result.data['user_id'] == 1
    handler.uow.commit.assert_called_once()

🏛️ Architecture

The package implements a simplified CQRS pattern with hybrid outbox pattern for event delivery:

┌─────────────────────────────────────────────────────────────┐
│              API Layer (DRF/FastAPI)                         │
├─────────────────────────────────────────────────────────────┤
│            Commands    │    Queries                          │
├─────────────────────────────────────────────────────────────┤
│           Handlers (Business Logic)                          │
│                   ↓                                          │
│           EventCollector (In-Memory)                         │
├─────────────────────────────────────────────────────────────┤
│          Unit of Work & Repositories                         │
│                   ↓                                          │
│      [Database] ← ATOMIC TRANSACTION → [Outbox Events]      │
│                                              ↓               │
│                                   [Immediate Publish 99%]    │
│                                              ↓               │
│                                          Kafka               │
│                                                              │
│      [Background Worker] → Retry Failed Events (1%)         │
└─────────────────────────────────────────────────────────────┘

Guarantees:
✅ Zero event loss (atomic commit)
✅ Low latency (~10-15ms for 99% of events)
✅ Guaranteed delivery (background worker retries)

🤝 Contributing

This package is used across all Felesh microservices. Any changes should be:

1. Backward compatible when possible
2. Thoroughly tested
3. Documented with examples
4. Reviewed by the architecture team

📄 License

Proprietary - Felesh Project

🆘 Support

• Check examples in scaffold/examples/cqrs_patterns/
• Review documentation in /docs directory
• Start with INDEX.md for complete reference
• Check TROUBLESHOOTING.md for common issues