Дата: 2025-11-10
Версия: 1.0.0
Цель: Система классов для создания CRM, ERP и любых приложений
YAML определение
↓ (парсинг)
Python метаклассы
↓ (генерация)
Динамические классы
↓ (регистрация)
SQLAlchemy модели + Pydantic схемы + FastAPI endpoints
↓ (runtime)
Работающее приложение
┌─────────────────────────────────────────────────────────┐
│ BASE CLASSES │
├─────────────────────────────────────────────────────────┤
│ │
│ BaseField (базовый класс для полей) │
│ ├── StringField │
│ ├── IntegerField │
│ ├── EmailField (extends StringField) │
│ ├── DateTimeField │
│ └── RelationField │
│ │
│ BaseEntity (базовый класс для сущностей) │
│ ├── id: UUIDField │
│ ├── created_at: DateTimeField │
│ ├── updated_at: DateTimeField │
│ └── методы: save(), delete(), validate() │
│ │
│ BaseModule (базовый класс для модулей) │
│ ├── entities: Dict[str, Type[BaseEntity]] │
│ ├── api_router: APIRouter │
│ └── методы: register(), get_entity() │
│ │
│ BaseComponent (базовый класс для компонентов) │
│ ├── modules: Dict[str, BaseModule] │
│ ├── config: Dict[str, Any] │
│ └── методы: initialize(), get_module() │
│ │
└─────────────────────────────────────────────────────────┘
# platform/core/metaclasses.py
from typing import Dict, Type, Any
class EntityMeta(type):
"""
Метакласс для автоматического создания Entity классов.
Делает:
1. Регистрирует класс в глобальном реестре
2. Создаёт SQLAlchemy модель
3. Создаёт Pydantic схемы (Create, Read, Update)
4. Создаёт API endpoints
5. Генерирует миграции
"""
# Глобальный реестр всех Entity
_registry: Dict[str, Type['BaseEntity']] = {}
def __new__(mcs, name, bases, namespace, **kwargs):
"""Вызывается при создании КЛАССА (не экземпляра!)"""
# Создаём класс
cls = super().__new__(mcs, name, bases, namespace)
# Пропускаем базовый класс
if name == 'BaseEntity':
return cls
# 1. Регистрируем в реестре
mcs._registry[name] = cls
# 2. Собираем все поля
fields = {}
for attr_name, attr_value in namespace.items():
if isinstance(attr_value, BaseField):
fields[attr_name] = attr_value
cls._fields = fields
# 3. Создаём SQLAlchemy модель
cls._sqlalchemy_model = mcs._create_sqlalchemy_model(cls, fields)
# 4. Создаём Pydantic схемы
cls._pydantic_create_schema = mcs._create_pydantic_schema(cls, fields, 'create')
cls._pydantic_read_schema = mcs._create_pydantic_schema(cls, fields, 'read')
cls._pydantic_update_schema = mcs._create_pydantic_schema(cls, fields, 'update')
# 5. Создаём API router
cls._api_router = mcs._create_api_router(cls)
print(f"✅ Entity '{name}' зарегистрирован с {len(fields)} полями")
return cls
@staticmethod
def _create_sqlalchemy_model(cls, fields):
"""Создаёт SQLAlchemy модель из полей"""
from sqlalchemy import Column, String, Integer, DateTime
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
# Динамически создаём атрибуты класса
attrs = {
'__tablename__': cls.__name__.lower() + 's',
'__table_args__': {'extend_existing': True},
}
# Добавляем поля
for field_name, field in fields.items():
attrs[field_name] = field.to_sqlalchemy_column()
# Создаём класс динамически
model_class = type(
cls.__name__ + 'Model',
(Base,),
attrs
)
return model_class
@staticmethod
def _create_pydantic_schema(cls, fields, mode: str):
"""Создаёт Pydantic схему для валидации"""
from pydantic import BaseModel, create_model
# Определяем какие поля включать
schema_fields = {}
for field_name, field in fields.items():
if mode == 'create' and field.read_only:
continue # Пропускаем read_only при создании
if mode == 'update' and field.immutable:
continue # Пропускаем immutable при обновлении
schema_fields[field_name] = field.to_pydantic_field()
# Динамически создаём Pydantic модель
schema_class = create_model(
cls.__name__ + mode.capitalize() + 'Schema',
**schema_fields
)
return schema_class
@staticmethod
def _create_api_router(cls):
"""Создаёт FastAPI router с CRUD endpoints"""
from fastapi import APIRouter, Depends, HTTPException
router = APIRouter(
prefix=f"/{cls.__name__.lower()}s",
tags=[cls.__name__]
)
# GET /entities (list)
@router.get("/", response_model=list[cls._pydantic_read_schema])
async def list_entities(skip: int = 0, limit: int = 100):
# TODO: query database
return []
# GET /entities/{id} (read)
@router.get("/{entity_id}", response_model=cls._pydantic_read_schema)
async def get_entity(entity_id: str):
# TODO: query database
raise HTTPException(404, "Not found")
# POST /entities (create)
@router.post("/", response_model=cls._pydantic_read_schema)
async def create_entity(data: cls._pydantic_create_schema):
# TODO: save to database
return data
# PUT /entities/{id} (update)
@router.put("/{entity_id}", response_model=cls._pydantic_read_schema)
async def update_entity(entity_id: str, data: cls._pydantic_update_schema):
# TODO: update in database
return data
# DELETE /entities/{id} (delete)
@router.delete("/{entity_id}")
async def delete_entity(entity_id: str):
# TODO: delete from database
return {"status": "deleted"}
return router
# platform/core/base.py
from platform.core.metaclasses import EntityMeta
class BaseEntity(metaclass=EntityMeta):
"""
Базовый класс для всех сущностей.
При наследовании автоматически:
- Регистрируется в реестре
- Создаётся SQLAlchemy модель
- Создаются Pydantic схемы
- Создаются API endpoints
"""
# Базовые поля для всех сущностей
id: 'UUIDField'
created_at: 'DateTimeField'
updated_at: 'DateTimeField'
def save(self):
"""Сохранить в БД"""
# Используем созданную SQLAlchemy модель
pass
def delete(self):
"""Удалить из БД"""
pass
def validate(self):
"""Валидировать через Pydantic схему"""
pass
# platform/core/fields.py
from typing import Any, Optional, Type
from sqlalchemy import Column, String as SQLString, Integer as SQLInteger
class BaseField:
"""
Базовый класс для полей.
Использует Descriptor Protocol для доступа к значениям:
- __get__ - чтение значения
- __set__ - запись значения
- __delete__ - удаление значения
"""
def __init__(
self,
*,
required: bool = True,
default: Any = None,
unique: bool = False,
index: bool = False,
read_only: bool = False,
immutable: bool = False,
validators: list = None,
help_text: str = "",
):
self.required = required
self.default = default
self.unique = unique
self.index = index
self.read_only = read_only
self.immutable = immutable
self.validators = validators or []
self.help_text = help_text
self.name: Optional[str] = None # Будет установлено метаклассом
def __set_name__(self, owner, name):
"""Вызывается когда поле присваивается атрибуту класса"""
self.name = name
def __get__(self, instance, owner):
"""Чтение значения"""
if instance is None:
return self # Если обращаются к классу, а не экземпляру
# Достаём значение из __dict__ экземпляра
return instance.__dict__.get(self.name)
def __set__(self, instance, value):
"""Запись значения"""
# Валидация
if self.immutable and self.name in instance.__dict__:
raise ValueError(f"Field '{self.name}' is immutable")
if self.read_only:
raise ValueError(f"Field '{self.name}' is read-only")
# Запускаем валидаторы
for validator in self.validators:
validator(value)
# Сохраняем значение
instance.__dict__[self.name] = value
def __delete__(self, instance):
"""Удаление значения"""
if self.name in instance.__dict__:
del instance.__dict__[self.name]
def to_sqlalchemy_column(self):
"""Конвертирует поле в SQLAlchemy Column"""
raise NotImplementedError("Subclasses must implement this")
def to_pydantic_field(self):
"""Конвертирует поле в Pydantic field annotation"""
raise NotImplementedError("Subclasses must implement this")
class StringField(BaseField):
"""Строковое поле"""
def __init__(self, max_length: int = 255, **kwargs):
super().__init__(**kwargs)
self.max_length = max_length
def __set__(self, instance, value):
# Дополнительная валидация для строк
if value and len(value) > self.max_length:
raise ValueError(f"String too long (max {self.max_length})")
super().__set__(instance, value)
def to_sqlalchemy_column(self):
from sqlalchemy import Column, String
return Column(
String(self.max_length),
nullable=not self.required,
unique=self.unique,
index=self.index,
default=self.default,
)
def to_pydantic_field(self):
from pydantic import Field
return (
str,
Field(
default=self.default,
max_length=self.max_length,
description=self.help_text,
)
)
class EmailField(StringField):
"""Email поле (расширяет StringField)"""
def __init__(self, **kwargs):
# Email всегда уникален и с валидатором
super().__init__(
max_length=255,
unique=True,
index=True,
validators=[self._validate_email],
**kwargs
)
@staticmethod
def _validate_email(value: str):
"""Валидатор email"""
import re
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
if not re.match(pattern, value):
raise ValueError(f"Invalid email: {value}")
def to_pydantic_field(self):
from pydantic import EmailStr, Field
return (
EmailStr,
Field(
default=self.default,
description=self.help_text,
)
)
class IntegerField(BaseField):
"""Целочисленное поле"""
def __init__(self, min_value: int = None, max_value: int = None, **kwargs):
super().__init__(**kwargs)
self.min_value = min_value
self.max_value = max_value
def __set__(self, instance, value):
# Валидация диапазона
if self.min_value is not None and value < self.min_value:
raise ValueError(f"Value too small (min {self.min_value})")
if self.max_value is not None and value > self.max_value:
raise ValueError(f"Value too large (max {self.max_value})")
super().__set__(instance, value)
def to_sqlalchemy_column(self):
from sqlalchemy import Column, Integer
return Column(
Integer,
nullable=not self.required,
unique=self.unique,
index=self.index,
default=self.default,
)
def to_pydantic_field(self):
from pydantic import Field
return (
int,
Field(
default=self.default,
ge=self.min_value,
le=self.max_value,
description=self.help_text,
)
)
class RelationField(BaseField):
"""Поле связи с другой сущностью"""
def __init__(self, target: str, many: bool = False, **kwargs):
super().__init__(**kwargs)
self.target = target # Имя целевой Entity
self.many = many # One-to-Many или One-to-One
def to_sqlalchemy_column(self):
from sqlalchemy import Column, ForeignKey, Integer
from sqlalchemy.orm import relationship
if self.many:
# Many-to-Many или One-to-Many
# Требует relationship, а не Column
return relationship(self.target)
else:
# Foreign Key
return Column(
Integer,
ForeignKey(f"{self.target.lower()}s.id"),
nullable=not self.required,
)
def to_pydantic_field(self):
from typing import Optional, List
# Динамически получаем целевую схему
target_schema = EntityMeta._registry[self.target]._pydantic_read_schema
if self.many:
return (List[target_schema], [])
else:
return (Optional[target_schema], None)
# platform/core/factory.py
from typing import Type, Dict, Any
from platform.core.base import BaseEntity
from platform.core.fields import (
StringField, IntegerField, EmailField,
DateTimeField, RelationField
)
import yaml
class EntityFactory:
"""
Фабрика для создания Entity классов из YAML определений.
"""
# Маппинг типов YAML → Python Field классов
FIELD_TYPE_MAP = {
'string': StringField,
'integer': IntegerField,
'email': EmailField,
'datetime': DateTimeField,
'relation': RelationField,
}
@classmethod
def from_yaml(cls, yaml_path: str) -> Type[BaseEntity]:
"""
Создаёт Entity класс из YAML файла.
Example:
Customer = EntityFactory.from_yaml('entities/customer.yaml')
"""
with open(yaml_path, 'r') as f:
config = yaml.safe_load(f)
return cls.from_dict(config)
@classmethod
def from_dict(cls, config: Dict[str, Any]) -> Type[BaseEntity]:
"""
Создаёт Entity класс из словаря.
Args:
config: {
'entity': 'Customer',
'fields': {
'name': {'type': 'string', 'max_length': 200},
'email': {'type': 'email', 'unique': True},
'age': {'type': 'integer', 'min_value': 0},
}
}
Returns:
Динамически созданный класс Customer(BaseEntity)
"""
entity_name = config['entity']
fields_config = config.get('fields', {})
# Создаём атрибуты класса
class_attrs = {}
for field_name, field_config in fields_config.items():
# Получаем класс поля
field_type = field_config.pop('type')
field_class = cls.FIELD_TYPE_MAP[field_type]
# Создаём экземпляр поля
field_instance = field_class(**field_config)
# Добавляем в атрибуты
class_attrs[field_name] = field_instance
# Динамически создаём класс
entity_class = type(
entity_name,
(BaseEntity,),
class_attrs
)
return entity_class
# Пример использования:
# 1. YAML файл: entities/customer.yaml
"""
entity: Customer
fields:
name:
type: string
max_length: 200
required: true
email:
type: email
unique: true
age:
type: integer
min_value: 0
max_value: 150
required: false
company:
type: relation
target: Company
many: false
"""
# 2. Создаём класс из YAML
Customer = EntityFactory.from_yaml('entities/customer.yaml')
# 3. Класс автоматически имеет:
# - SQLAlchemy модель (Customer._sqlalchemy_model)
# - Pydantic схемы (Customer._pydantic_create_schema, etc.)
# - API router (Customer._api_router)
# 4. Используем класс
customer = Customer()
customer.name = "Иван Петров"
customer.email = "ivan@example.com"
customer.age = 30
customer.save() # Сохраняет в БД
# 5. API автоматически доступно:
# POST /customers → создать
# GET /customers → список
# GET /customers/{id} → получить
# PUT /customers/{id} → обновить
# DELETE /customers/{id} → удалить
# platform/core/mixins.py
class TimestampMixin:
"""Добавляет автоматические timestamps"""
created_at = DateTimeField(auto_now_add=True, read_only=True)
updated_at = DateTimeField(auto_now=True, read_only=True)
class SoftDeleteMixin:
"""Мягкое удаление (не удаляет из БД, а помечает)"""
deleted_at = DateTimeField(nullable=True, default=None)
is_deleted = BooleanField(default=False)
def delete(self):
"""Переопределяем delete для мягкого удаления"""
from datetime import datetime
self.deleted_at = datetime.now()
self.is_deleted = True
self.save()
def restore(self):
"""Восстановить удалённый объект"""
self.deleted_at = None
self.is_deleted = False
self.save()
class OwnershipMixin:
"""Добавляет владельца записи"""
owner = RelationField(target='User', required=True)
def can_edit(self, user):
"""Проверка прав на редактирование"""
return self.owner.id == user.id or user.is_admin
class AuditMixin:
"""Аудит изменений"""
created_by = RelationField(target='User', read_only=True)
updated_by = RelationField(target='User')
def save(self, user):
"""Переопределяем save для записи истории"""
if not self.id: # Новый объект
self.created_by = user
self.updated_by = user
# Сохраняем в audit log
AuditLog.create(
entity_type=self.__class__.__name__,
entity_id=self.id,
action='create' if not self.id else 'update',
user=user,
changes=self._get_changes()
)
super().save()
class SearchableMixin:
"""Полнотекстовый поиск"""
@classmethod
def search(cls, query: str, limit: int = 100):
"""Поиск по всем текстовым полям"""
# Собираем все текстовые поля
text_fields = [
name for name, field in cls._fields.items()
if isinstance(field, StringField)
]
# TODO: SQL запрос с LIKE или full-text search
pass
# Использование миксинов:
class Customer(
TimestampMixin, # Автоматические timestamps
SoftDeleteMixin, # Мягкое удаление
OwnershipMixin, # Владелец записи
AuditMixin, # Аудит изменений
SearchableMixin, # Поиск
BaseEntity # Базовая сущность (всегда последний!)
):
name = StringField(max_length=200)
email = EmailField()
# Теперь Customer автоматически имеет:
# - created_at, updated_at (от TimestampMixin)
# - deleted_at, is_deleted, restore() (от SoftDeleteMixin)
# - owner, can_edit() (от OwnershipMixin)
# - created_by, updated_by (от AuditMixin)
# - search() (от SearchableMixin)
# crm/entities/contact.yaml
entity: Contact
extends: [TimestampMixin, OwnershipMixin, SearchableMixin]
fields:
# Основные данные
first_name:
type: string
max_length: 100
required: true
last_name:
type: string
max_length: 100
required: true
email:
type: email
unique: true
required: true
phone:
type: phone
required: false
# Связи
company:
type: relation
target: Company
many: false
required: false
deals:
type: relation
target: Deal
many: true
# Дополнительно
position:
type: string
max_length: 200
required: false
status:
type: enum
values: [lead, customer, partner, inactive]
default: lead
tags:
type: tags
max_tags: 10
# Методы (custom logic)
methods:
full_name:
type: property
code: "return f'{self.first_name} {self.last_name}'"
send_email:
type: method
params:
subject: string
body: string
code: |
from platform.services.email import EmailService
EmailService.send(
to=self.email,
subject=subject,
body=body
)
# API расширения
api:
custom_endpoints:
- path: /{id}/send_email
method: POST
handler: send_email
auth_required: true
# crm/entities/company.yaml
entity: Company
extends: [TimestampMixin, OwnershipMixin, SearchableMixin]
fields:
name:
type: string
max_length: 300
required: true
unique: true
website:
type: url
required: false
industry:
type: enum
values: [tech, finance, retail, manufacturing, other]
size:
type: enum
values: [startup, small, medium, large, enterprise]
contacts:
type: relation
target: Contact
many: true
deals:
type: relation
target: Deal
many: true
# crm/entities/deal.yaml
entity: Deal
extends: [TimestampMixin, OwnershipMixin, AuditMixin]
fields:
title:
type: string
max_length: 300
required: true
amount:
type: decimal
min_value: 0
required: true
currency:
type: enum
values: [RUB, USD, EUR]
default: RUB
stage:
type: enum
values: [lead, qualification, proposal, negotiation, won, lost]
default: lead
probability:
type: integer
min_value: 0
max_value: 100
default: 50
expected_close_date:
type: date
required: false
# Связи
contact:
type: relation
target: Contact
required: true
company:
type: relation
target: Company
required: false
# Методы
methods:
mark_as_won:
type: method
code: |
self.stage = 'won'
self.probability = 100
self.save()
# Создать invoice автоматически
from crm.entities import Invoice
Invoice.create_from_deal(self)
mark_as_lost:
type: method
params:
reason: string
code: |
self.stage = 'lost'
self.probability = 0
self.lost_reason = reason
self.save()
# crm/generate.py
from platform.core.factory import EntityFactory
from pathlib import Path
# Создаём классы из YAML
crm_entities = {}
for yaml_file in Path('crm/entities/').glob('*.yaml'):
entity_name = yaml_file.stem.capitalize()
entity_class = EntityFactory.from_yaml(yaml_file)
crm_entities[entity_name] = entity_class
Contact = crm_entities['Contact']
Company = crm_entities['Company']
Deal = crm_entities['Deal']
# Теперь CRM работает!
# Создаём контакт
contact = Contact()
contact.first_name = "Иван"
contact.last_name = "Петров"
contact.email = "ivan@example.com"
contact.phone = "+7 (999) 123-45-67"
contact.status = "lead"
contact.save()
# Создаём компанию
company = Company()
company.name = "ООО Рога и Копыта"
company.industry = "retail"
company.size = "medium"
company.save()
# Связываем контакт с компанией
contact.company = company
contact.save()
# Создаём сделку
deal = Deal()
deal.title = "Продажа системы CRM"
deal.amount = 500000
deal.currency = "RUB"
deal.stage = "proposal"
deal.contact = contact
deal.company = company
deal.save()
# Закрываем сделку
deal.mark_as_won()
# API автоматически работает:
# GET /contacts → список контактов
# POST /contacts → создать контакт
# GET /contacts/{id} → получить контакт
# PUT /contacts/{id} → обновить
# DELETE /contacts/{id} → удалить
# POST /contacts/{id}/send_email → отправить email
# GET /companies → список компаний
# POST /deals → создать сделку
# etc...
# erp/entities/product.yaml
entity: Product
extends: [TimestampMixin, AuditMixin]
fields:
sku:
type: string
max_length: 50
unique: true
required: true
name:
type: string
max_length: 300
required: true
description:
type: text
category:
type: relation
target: ProductCategory
price:
type: decimal
min_value: 0
cost:
type: decimal
min_value: 0
stock:
type: integer
min_value: 0
default: 0
reorder_level:
type: integer
min_value: 0
default: 10
methods:
needs_reorder:
type: property
code: "return self.stock <= self.reorder_level"
adjust_stock:
type: method
params:
quantity: integer
reason: string
code: |
old_stock = self.stock
self.stock += quantity
self.save()
# Записываем в историю движения
StockMovement.create(
product=self,
quantity=quantity,
old_stock=old_stock,
new_stock=self.stock,
reason=reason
)
# erp/entities/purchase_order.yaml
entity: PurchaseOrder
extends: [TimestampMixin, OwnershipMixin, AuditMixin]
fields:
po_number:
type: string
max_length: 50
unique: true
auto_generate: "PO-{YYYY}-{counter}"
supplier:
type: relation
target: Supplier
required: true
status:
type: enum
values: [draft, sent, confirmed, received, cancelled]
default: draft
items:
type: relation
target: PurchaseOrderItem
many: true
total:
type: decimal
min_value: 0
read_only: true # Вычисляется автоматически
expected_delivery_date:
type: date
actual_delivery_date:
type: date
nullable: true
methods:
calculate_total:
type: method
code: |
total = sum(item.subtotal for item in self.items)
self.total = total
self.save()
confirm:
type: method
code: |
self.status = 'confirmed'
self.save()
# Отправить email поставщику
self.supplier.send_email(
subject=f"Purchase Order {self.po_number}",
body=f"Please confirm PO {self.po_number}"
)
receive:
type: method
code: |
from datetime import date
self.status = 'received'
self.actual_delivery_date = date.today()
self.save()
# Обновить склад
for item in self.items:
item.product.adjust_stock(
quantity=item.quantity,
reason=f"Received PO {self.po_number}"
)
# erp/entities/invoice.yaml
entity: Invoice
extends: [TimestampMixin, AuditMixin]
fields:
invoice_number:
type: string
max_length: 50
unique: true
auto_generate: "INV-{YYYY}-{counter}"
customer:
type: relation
target: Customer
required: true
status:
type: enum
values: [draft, sent, paid, overdue, cancelled]
default: draft
items:
type: relation
target: InvoiceItem
many: true
subtotal:
type: decimal
read_only: true
tax:
type: decimal
read_only: true
total:
type: decimal
read_only: true
due_date:
type: date
paid_date:
type: date
nullable: true
methods:
calculate_totals:
type: method
code: |
self.subtotal = sum(item.subtotal for item in self.items)
self.tax = self.subtotal * 0.20 # 20% НДС
self.total = self.subtotal + self.tax
self.save()
mark_as_paid:
type: method
code: |
from datetime import date
self.status = 'paid'
self.paid_date = date.today()
self.save()
# Записать в бухгалтерию
AccountingEntry.create_from_invoice(self)
# platform/core/module.py
from typing import Dict, Type, List
from fastapi import APIRouter
class BaseModule:
"""
Базовый класс для модулей.
Модуль = набор Entity + API + логика
"""
def __init__(self, config: Dict):
self.config = config
self.entities: Dict[str, Type[BaseEntity]] = {}
self.router = APIRouter()
def register_entity(self, entity_class: Type[BaseEntity]):
"""Регистрирует Entity в модуле"""
entity_name = entity_class.__name__
self.entities[entity_name] = entity_class
# Добавляем API router сущности к модулю
self.router.include_router(
entity_class._api_router,
prefix=f"/{self.config.get('api_prefix', '')}"
)
def get_entity(self, name: str) -> Type[BaseEntity]:
"""Получить Entity по имени"""
return self.entities.get(name)
def initialize(self):
"""Инициализация модуля (вызывается при старте)"""
pass
class ModuleFactory:
"""Фабрика для создания модулей из YAML"""
@classmethod
def from_yaml(cls, yaml_path: str) -> BaseModule:
"""Создаёт модуль из YAML"""
with open(yaml_path, 'r') as f:
config = yaml.safe_load(f)
module = BaseModule(config)
# Создаём все Entity из конфигурации
for entity_config in config.get('entities', []):
entity_class = EntityFactory.from_dict(entity_config)
module.register_entity(entity_class)
return module
# Пример: CRM модуль
# crm/crm_module.yaml
"""
module:
name: CRMModule
version: 1.0.0
api_prefix: /crm
entities:
- entity: Contact
fields: {...}
- entity: Company
fields: {...}
- entity: Deal
fields: {...}
config:
default_currency: RUB
enable_email_tracking: true
"""
# Создаём модуль
crm_module = ModuleFactory.from_yaml('crm/crm_module.yaml')
# Получаем Entity из модуля
Contact = crm_module.get_entity('Contact')
Company = crm_module.get_entity('Company')
Deal = crm_module.get_entity('Deal')
# API модуля
app.include_router(crm_module.router) # /crm/contacts, /crm/companies, /crm/deals
# platform/core/component.py
class BaseComponent:
"""
Базовый класс для компонентов.
Компонент = набор Модулей + общая конфигурация
"""
def __init__(self, config: Dict):
self.config = config
self.modules: Dict[str, BaseModule] = {}
self.router = APIRouter()
def register_module(self, name: str, module: BaseModule):
"""Регистрирует модуль в компоненте"""
self.modules[name] = module
# Добавляем API router модуля к компоненту
self.router.include_router(
module.router,
prefix=f"/{self.config.get('component_prefix', name)}"
)
def get_module(self, name: str) -> BaseModule:
"""Получить модуль по имени"""
return self.modules.get(name)
def initialize(self):
"""Инициализация компонента"""
for module in self.modules.values():
module.initialize()
# Пример: ERP компонент
class ERPComponent(BaseComponent):
def __init__(self, config: Dict):
super().__init__(config)
# Регистрируем модули ERP
self.register_module('inventory', InventoryModule(config))
self.register_module('purchasing', PurchasingModule(config))
self.register_module('sales', SalesModule(config))
self.register_module('accounting', AccountingModule(config))
# Использование
erp = ERPComponent({
'component_prefix': 'erp',
'default_currency': 'RUB',
'enable_multi_warehouse': True,
})
erp.initialize()
# API компонента
app.include_router(erp.router) # /erp/inventory/*, /erp/purchasing/*, etc.
# crm_app.yaml
component:
name: CRMComponent
version: 1.0.0
modules:
contacts:
entities:
Contact:
extends: [TimestampMixin, OwnershipMixin, SearchableMixin]
fields:
first_name: {type: string, max_length: 100, required: true}
last_name: {type: string, max_length: 100, required: true}
email: {type: email, unique: true, required: true}
phone: {type: phone}
company: {type: relation, target: Company}
companies:
entities:
Company:
extends: [TimestampMixin, OwnershipMixin]
fields:
name: {type: string, max_length: 300, unique: true, required: true}
website: {type: url}
industry: {type: enum, values: [tech, finance, retail]}
deals:
entities:
Deal:
extends: [TimestampMixin, OwnershipMixin, AuditMixin]
fields:
title: {type: string, max_length: 300, required: true}
amount: {type: decimal, min_value: 0, required: true}
stage: {type: enum, values: [lead, proposal, won, lost]}
contact: {type: relation, target: Contact, required: true}
$ python -m platform.cli generate crm_app.yaml
🔄 Generating CRM application...
✅ Created entities:
- Contact (3 modules: contacts)
- Company (1 module: companies)
- Deal (1 module: deals)
✅ Created SQLAlchemy models:
- generated/crm/models/contact.py
- generated/crm/models/company.py
- generated/crm/models/deal.py
✅ Created Pydantic schemas:
- generated/crm/schemas/contact.py
- generated/crm/schemas/company.py
- generated/crm/schemas/deal.py
✅ Created API routers:
- generated/crm/api/contacts.py
- generated/crm/api/companies.py
- generated/crm/api/deals.py
✅ Created database migrations:
- generated/crm/migrations/001_initial.py
✅ Created FastAPI app:
- generated/crm/main.py
Done! Run: cd generated/crm && uvicorn main:app
# generated/crm/main.py (автоматически создан!)
from fastapi import FastAPI
from platform.core.factory import EntityFactory, ModuleFactory, ComponentFactory
# Создаём приложение
app = FastAPI(title="CRM Application", version="1.0.0")
# Создаём компонент из YAML
crm_component = ComponentFactory.from_yaml('crm_app.yaml')
# Инициализируем
crm_component.initialize()
# Подключаем API
app.include_router(crm_component.router)
# Готово! CRM работает.
$ cd generated/crm
$ uvicorn main:app --reload
INFO: Started server process
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8000
✅ CRM работает!
API доступен:
- GET /contacts → список контактов
- POST /contacts → создать контакт
- GET /contacts/{id} → получить контакт
- PUT /contacts/{id} → обновить
- DELETE /contacts/{id} → удалить
- GET /companies → список компаний
- POST /companies → создать компанию
- GET /deals → список сделок
- POST /deals → создать сделку
Swagger UI: http://127.0.0.1:8000/docs
1️⃣ МЕТАПРОГРАММИРОВАНИЕ
- Метаклассы создают классы автоматически
- Регистрация в реестре
- Автоматическое создание SQLAlchemy + Pydantic + FastAPI
2️⃣ DESCRIPTOR PROTOCOL
- Поля как объекты с валидацией
- Автоматическая конвертация типов
- Расширяемая система полей
3️⃣ FACTORY PATTERN
- YAML → Python классы автоматически
- Динамическое создание классов
- Нулевой boilerplate код
4️⃣ MIXIN SYSTEM
- Переиспользуемый функционал
- Композиция вместо наследования
- Timestamps, SoftDelete, Ownership, Audit
5️⃣ МОДУЛЬНАЯ СИСТЕМА
- Модуль = набор Entity + API
- Компонент = набор Модулей
- Изоляция и переиспользование
6️⃣ ПОЛНЫЕ ПРИЛОЖЕНИЯ
- CRM из 50 строк YAML → 2000+ строк кода
- ERP из конфигурации
- Любое приложение за минуты
Готово к имплементации! 🚀