pyline-core 2.2.0

pyline core

PyLine Core

A lightweight Python framework for implementing the Command Query Responsibility Segregation (CQRS) pattern with pipeline orchestration capabilities.

Overview

PyLine Core provides a clean architecture for building applications using the CQRS pattern, where commands and queries are separated, and handlers are managed through a mediator pattern. The framework also includes a pipeline system for orchestrating complex workflows.

Features

• Command Pattern: Separate command objects from their handlers
• Query Pattern: Dedicated query objects with result types
• Mediator Pattern: Centralized handler registration and execution
• Pipeline Orchestration: Chain commands and queries in sequential workflows
• Async Support: Native asyncio support for high-performance I/O operations
• Type Safety: Built with Python type hints for better IDE support
• Minimal Dependencies: Lightweight with no external dependencies

Installation

pip install pyline-core

Quick Start

1. Define Commands and Handlers

from pyline import Command, CommandHandler
from dataclasses import dataclass

@dataclass
class CreateUserCommand(Command):
    name: str

class CreateUserCommandHandler(CommandHandler):
    async def handle(self, command: CreateUserCommand):
        print(f"Creating user: {command.name}")
        # Your business logic here

2. Define Queries and Handlers

from pyline import Query, QueryResult, QueryHandler
from dataclasses import dataclass

@dataclass
class GetUserByNameQuery(Query):
    name: str

@dataclass
class GetUserByNameQueryResult(QueryResult):
    user: dict
    email: str

class GetUserByNameQueryHandler(QueryHandler):
    async def handle(self, query: GetUserByNameQuery):
        # Your data access logic here
        return GetUserByNameQueryResult(
            user={"id": 1, "name": query.name, "email": "[e-mail]"},
            email="[e-mail]"
        )

3. Register Handlers

from pyline import mediator

mediator.register_handler(CreateUserCommand, CreateUserCommandHandler())
mediator.register_handler(GetUserByNameQuery, GetUserByNameQueryHandler())

4. Execute Commands and Queries

import asyncio

async def main():
    # Execute a command
    command = CreateUserCommand(name="John Doe")
    await mediator.send(command)

    # Execute a query
    query = GetUserByNameQuery(name="John Doe")
    result = await mediator.send(query)
    print(f"User email: {result.email}")

if __name__ == "__main__":
    asyncio.run(main())

Pipeline Orchestration

PyLine Core includes a powerful pipeline system for orchestrating complex workflows:

from pyline.pipe import Pipe

# Define a pipeline
create_user_pipe = Pipe(
    name="Create User Pipeline",
    context={
        "name": "John Doe",
    },
    steps=[
        CreateUserCommand,
        GetUserByNameQuery,
        # Add more commands/queries as needed
    ],
)

# Execute the pipeline
async def main():
    await create_user_pipe.run()

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

Pipeline Features

• Context Sharing: Data flows between pipeline steps through a shared context
• Automatic Parameter Mapping: Pipeline automatically maps context data to command/query parameters
• Result Propagation: Query results are automatically added to the context for subsequent steps
• Step Tracking: Built-in logging shows progress through pipeline execution

Architecture

Core Components

1. Command: Abstract base class for commands
2. CommandHandler: Abstract base class for command handlers
3. Query: Abstract base class for queries
4. QueryResult: Abstract base class for query results
5. QueryHandler: Abstract base class for query handlers
6. HandlerMediator: Central registry and dispatcher for handlers
7. Pipe: Pipeline orchestration system

Design Patterns

• Command Pattern: Encapsulates requests as objects
• Query Pattern: Separates read operations from write operations
• Mediator Pattern: Centralizes communication between components
• Pipeline Pattern: Orchestrates sequential execution of operations

Architecture Flow

graph TD
    Client([Client / API]) -->|Starts Workflow| Context[{Shared Context}]
    Client -->|Triggers| P[Pipe Orchestrator]
    
    P --> |Step 1: Extracts Params| C[Command / Query]
    P --> |Step 2: Sends to Mediator| M{Handler Mediator}
    
    M --> |Resolves Handler| H[CommandHandler / QueryHandler]
    H --> |Executes Logic| DB[(Data Layer)]
    
    H -.-> |Returns Result| P
    P -.-> |Updates Context| Context
    
    P --> |Next Step| C2[Next Command / Query]

Requirements

• Python 3.10+

License

This project is licensed under the terms specified in the LICENSE file.

Contributing

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

Author

Alp Sakaci
Email: [e-mail]