pyd4all 2024.12.20.2
All the pydantic projects I want to use together.
Requires Python: <4.0,>=3.12

# pydantic-all-in-one

[![Dev Containers](https://img.shields.io/static/v1?label=Dev%20Containers&message=Open&color=blue&logo=visualstudiocode)](https://vscode.dev/redirect?url=vscode://ms-vscode-remote.remote-containers/cloneInVolume?url=https://github.com/seanchatmangpt/pydantic-all-in-one) [![GitHub Codespaces](https://img.shields.io/static/v1?label=GitHub%20Codespaces&message=Open&color=blue&logo=github)](https://github.com/codespaces/new/seanchatmangpt/pydantic-all-in-one)

```sh
pip install pydantic-all-in-one
```

## Overview

**pydantic-all-in-one** is a unified Python framework that seamlessly integrates multiple Pydantic-based libraries to provide a comprehensive solution for building scalable, efficient, and compliant applications. Over five years of development, it has evolved into a robust ecosystem used by Fortune 10 companies to ensure compliance with the EU AI Act through the innovative concept of **Service Colonies**.

Key features include:

- **Intelligent Model Generation with DSLModel**: Create dynamic models using templates and AI-assisted code generation.
- **EU AI Act Compliance**: Implement features to ensure AI applications meet EU regulatory standards.
- **Service Colony Architecture**: Develop autonomous and cooperative services that adapt and evolve.
- **FastAPI Integration**: Build high-performance RESTful APIs.
- **SQLModel and GQLAlchemy**: Manage relational and graph databases seamlessly.
- **FastStream and Redis**: Implement event-driven architectures.
- **aioclock for Scheduling**: Schedule and manage asynchronous tasks.
- **LanceDB**: Handle vector data for machine learning applications.
- **Typer for CLI**: Create powerful command-line interfaces.
- **Extensive Testing and CI/CD**: Ensure code quality with comprehensive testing and continuous integration pipelines.

## Table of Contents

- [pydantic-all-in-one](#pydantic-all-in-one)
  - [Overview](#overview)
  - [Table of Contents](#table-of-contents)
  - [Installation](#installation)
  - [Getting Started](#getting-started)
    - [Defining Intelligent Models with DSLModel](#defining-intelligent-models-with-dslmodel)
    - [Implementing Service Colonies for EU AI Act Compliance](#implementing-service-colonies-for-eu-ai-act-compliance)
    - [Dynamic Class Generation](#dynamic-class-generation)
    - [Workflow Management and Scheduling](#workflow-management-and-scheduling)
    - [Event-Driven Architecture with FastStream and Redis](#event-driven-architecture-with-faststream-and-redis)
    - [Graph and Relational Data Management](#graph-and-relational-data-management)
    - [Data Handling and Vector Management](#data-handling-and-vector-management)
    - [CLI Integration with Typer](#cli-integration-with-typer)
  - [Architecture](#architecture)
    - [Core Components](#core-components)
    - [Service Colony Architecture](#service-colony-architecture)
    - [Data Flow](#data-flow)
  - [Development](#development)
    - [Setup](#setup)
    - [Testing](#testing)
    - [Continuous Integration and Deployment](#continuous-integration-and-deployment)
  - [Contributing](#contributing)
  - [License](#license)
  - [Contact](#contact)

## Installation

Ensure you have Python 3.12 or higher installed. Then, install **pydantic-all-in-one** via pip:

```sh
pip install pydantic-all-in-one
```

Alternatively, install from source:

```sh
git clone https://github.com/seanchatmangpt/pydantic-all-in-one.git
cd pydantic-all-in-one
poetry install
```

## Getting Started

### Defining Intelligent Models with DSLModel

Leverage **DSLModel** to create dynamic, intelligent models using Jinja2 templates and AI-assisted code generation.

```python
# models.py

from typing import List, Optional
from pydantic import Field
from dslmodel import DSLModel, init_lm

# Initialize language model for AI-assisted generation
init_lm()  # Sets the language model to 'gpt-4o-mini'

class ComplianceRequirement(DSLModel):
    """Represents a compliance requirement under the EU AI Act."""
    requirement_id: str = Field(..., description="Unique identifier for the requirement.")
    description: str = Field(..., description="Description of the compliance requirement.")
    risk_level: str = Field(..., description="Risk level associated with the requirement.")

class AIComponent(DSLModel):
    """Represents an AI component within the system."""
    component_id: str = Field(..., description="Unique identifier for the AI component.")
    functionalities: List[str] = Field(..., description="List of functionalities.")
    compliance_status: str = Field(..., description="Compliance status with the EU AI Act.")
```

Generate models from templates:

```python
# generate_models.py

from models import AIComponent

component_template = """
Component ID: {{ uuid4() }}
Functionalities:
{% for func in functionalities %}
- {{ func }}
{% endfor %}
Compliance Status: Pending
"""

functionalities = ["Data Processing", "Automated Decision Making", "User Interaction"]

# Create an AIComponent instance using the template
ai_component = AIComponent.from_prompt(component_template, functionalities=functionalities)

print(ai_component.json(indent=2))
```

### Implementing Service Colonies for EU AI Act Compliance

Develop a **Service Colony** architecture where autonomous services (inhabitants) collaborate to ensure compliance with the EU AI Act.

```python
# service_colony.py

from dslmodel import FSMMixin, trigger
from enum import Enum, auto
from typing import List

class ComplianceState(Enum):
    INIT = auto()
    ASSESSING = auto()
    COMPLYING = auto()
    MONITORING = auto()

class ComplianceInhabitant(FSMMixin):
    def __init__(self, component: AIComponent):
        super().__init__()
        self.component = component
        self.setup_fsm(state_enum=ComplianceState, initial=ComplianceState.INIT)

    @trigger(source=ComplianceState.INIT, dest=ComplianceState.ASSESSING)
    def assess_risk(self):
        print(f"Assessing risk for {self.component.component_id}")
        # Perform risk assessment...

    @trigger(source=ComplianceState.ASSESSING, dest=ComplianceState.COMPLYING)
    def implement_compliance(self):
        print(f"Implementing compliance measures for {self.component.component_id}")
        # Implement compliance...

    @trigger(source=ComplianceState.COMPLYING, dest=ComplianceState.MONITORING)
    def start_monitoring(self):
        print(f"Starting monitoring for {self.component.component_id}")
        # Start monitoring...

    def forward(self, event: str):
        super().forward(event)
        print(f"Processing event: {event}")
```

Instantiate inhabitants and simulate compliance workflow:

```python
# main.py

from service_colony import ComplianceInhabitant
from models import AIComponent

component = AIComponent(
    component_id="comp-123",
    functionalities=["Automated Decision Making"],
    compliance_status="Pending"
)

inhabitant = ComplianceInhabitant(component)
inhabitant.assess_risk()
inhabitant.implement_compliance()
inhabitant.start_monitoring()
```

### Dynamic Class Generation

Use **DSLClassGenerator** for dynamic class creation based on prompts.

```python
# class_generator.py

from dslmodel.generators.gen_models import DSLClassGenerator
from pathlib import Path

# Generate a new Pydantic model class from a prompt
prompt = """
Create a Pydantic model class named 'RiskAssessment' with fields:
- assessment_id: str
- component_id: str
- risk_level: str
- findings: List[str]
- recommendations: List[str]
"""

generator = DSLClassGenerator(
    model_prompt=prompt,
    file_path=Path('./generated_models.py'),
    append=True
)

generator()
```

Generated class (`generated_models.py`):

```python
from typing import List
from pydantic import BaseModel

class RiskAssessment(BaseModel):
    assessment_id: str
    component_id: str
    risk_level: str
    findings: List[str]
    recommendations: List[str]
```

### Workflow Management and Scheduling

Define and execute complex workflows using **Workflow**, **Job**, and **Action**, and schedule tasks with **aioclock**.

```python
# workflow.py

from dslmodel.workflow import Workflow, Job, Action, Condition, CronSchedule
from aioclock import AioClock, Every

# Define actions
action_assess = Action(
    name="Assess Risk",
    code="inhabitant.assess_risk()"
)

action_implement = Action(
    name="Implement Compliance",
    code="inhabitant.implement_compliance()"
)

action_monitor = Action(
    name="Start Monitoring",
    code="inhabitant.start_monitoring()"
)

# Define job
job = Job(
    name="Compliance Workflow",
    steps=[action_assess, action_implement, action_monitor]
)

# Define workflow
workflow = Workflow(
    name="EU AI Act Compliance Workflow",
    jobs=[job],
    context={"inhabitant": inhabitant}
)

# Schedule workflow execution
clock = AioClock()

@clock.task(trigger=Every(hours=24))
async def scheduled_workflow():
    workflow.execute()
```

### Event-Driven Architecture with FastStream and Redis

Implement an event-driven architecture using **FastStream** and **Redis**.

```python
# event_stream.py

from faststream import FastStream
from faststream.redis import RedisBroker
from models import ComplianceRequirement

broker = RedisBroker("redis://localhost:6379")
app = FastStream(broker)

@app.subscriber("compliance/requirements")
async def handle_requirement(data: ComplianceRequirement):
    print(f"Received compliance requirement: {data.requirement_id}")
    # Process requirement...

@app.publisher("compliance/status")
async def publish_status(status: dict):
    await broker.publish("compliance/status", status)
```

### Graph and Relational Data Management

Manage graph data with **GQLAlchemy** and relational data with **SQLModel**.

```python
# data_management.py

from gqlalchemy import Memgraph, Node
from sqlmodel import SQLModel, Field, create_engine, Session

# Memgraph setup
memgraph = Memgraph()

class ComponentNode(Node):
    component_id: str
    compliance_status: str

# SQLModel setup
class ComplianceRecord(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True)
    component_id: str
    status: str

engine = create_engine("sqlite:///compliance.db")
SQLModel.metadata.create_all(engine)
```

### Data Handling and Vector Management

Use **DataReader** and **DataWriter** for data handling, and **LanceDB** for vector data management.

```python
# data_io.py

from dslmodel import DataReader, DataWriter
from lancedb.pydantic import pydantic_to_schema
from models import ComplianceRequirement

# Reading data
data_reader = DataReader(file_path="data/requirements.csv")
requirements = data_reader.forward()

# Writing data
data_writer = DataWriter(data=requirements, file_path="output/processed_requirements.csv")
data_writer.forward()

# LanceDB schema
requirement_schema = pydantic_to_schema(ComplianceRequirement)
```

### CLI Integration with Typer

Create powerful command-line interfaces using **Typer**.

```python
# cli.py

import typer

app = typer.Typer()

@app.command()
def assess(component_id: str):
    """Assess compliance for a component."""
    # Perform assessment...
    typer.echo(f"Compliance assessment started for {component_id}")

@app.command()
def status(component_id: str):
    """Check compliance status of a component."""
    # Check status...
    typer.echo(f"Compliance status for {component_id}: Compliant")

if __name__ == "__main__":
    app()
```

## Architecture

### Core Components

- **DSLModel**: Core framework for intelligent model creation using templates and AI assistance.
- **Service Colonies**: Architectural style for developing autonomous and cooperative services.
- **FSMMixin**: Provides finite state machine functionality.
- **Workflow Components**: `Workflow`, `Job`, `Action`, `Condition`, `CronSchedule` for orchestrating workflows.
- **Data Handling Utilities**: `DataReader`, `DataWriter` for data ingestion and output.
- **Database Management**: **SQLModel** and **GQLAlchemy** for relational and graph databases.
- **Event Streaming**: **FastStream** and **RedisBroker** for event-driven architectures.
- **Scheduling**: **aioclock** for scheduling asynchronous tasks.
- **Vector Management**: **LanceDB** for handling vector data.

### Service Colony Architecture

The Service Colony consists of inhabitants (services) that:

- **Autonomously adapt** to changes.
- **Collaborate** to fulfill global objectives.
- **Ensure compliance** with regulations like the EU AI Act.
- **Evolve** over time through dynamic class generation and AI assistance.

### Data Flow

```
User Inputs -> DSLModel Templates -> Generated Models -> Inhabitants (Services)
       |
       v
Event Streams (FastStream) <-> Inhabitants
       |
       v
Data Storage (SQLModel, GQLAlchemy, LanceDB)
       |
       v
Monitoring and Compliance Reporting
```

## Development

### Setup

1. **Clone the Repository**

   ```sh
   git clone https://github.com/seanchatmangpt/pydantic-all-in-one.git
   cd pydantic-all-in-one
   ```

2. **Install Dependencies**

   ```sh
   poetry install
   ```

3. **Configure Environment Variables**

   Create a `.env` file and add necessary environment variables, such as `REDIS_URL` and `MEMGRAPH_URL`.

4. **Start Docker Services**

   ```sh
   docker-compose up -d
   ```

5. **Run the Application**

   ```sh
   poetry run poe api
   ```

### Testing

Run tests using `pytest`:

```sh
poetry run pytest
```

Ensure test coverage is at least **90%**.

### Continuous Integration and Deployment

**pydantic-all-in-one** utilizes GitHub Actions for CI/CD:

- **Code Push**: Triggers automated testing and linting.
- **Testing**: Runs unit and integration tests.
- **Linting**: Uses `ruff` for code quality checks.
- **Deployment**: Deploys successful builds to staging or production environments.

## Contributing

Contributions are welcome! Please follow the [contribution guidelines](CONTRIBUTING.md) and adhere to the code of conduct.

1. **Fork the Repository**

2. **Create a Feature Branch**

   ```sh
   git checkout -b feature/YourFeature
   ```

3. **Commit Your Changes**

4. **Push to Your Fork**

   ```sh
   git push origin feature/YourFeature
   ```

5. **Open a Pull Request**

## License

Distributed under the MIT License. See [LICENSE](LICENSE) for more information.

## Contact

- **Project Link**: [https://github.com/seanchatmangpt/pydantic-all-in-one](https://github.com/seanchatmangpt/pydantic-all-in-one)
- **Issues**: [https://github.com/seanchatmangpt/pydantic-all-in-one/issues](https://github.com/seanchatmangpt/pydantic-all-in-one/issues)

---

By following this guide, you can effectively utilize **pydantic-all-in-one** to build scalable, efficient, and compliant applications. The integration of **DSLModel** provides intelligence and initiative, enabling the development of dynamic models and services that adapt over time. The Service Colony architecture fosters collaboration among autonomous services, ensuring compliance with regulations like the EU AI Act.

This comprehensive README illustrates how all the components and classes work together, providing a cohesive narrative and practical examples to get you started.

---

**Note**: The project structure reflects five years of development, incorporating advanced features and compliance mechanisms used by Fortune 10 companies. The examples demonstrate how to build a production-ready system that leverages modern Python frameworks and AI assistance to meet stringent regulatory requirements.

---

Happy coding!