🏗️ Архитектура системы¶
📐 Общая архитектура¶
Система охраны труда построена по принципу многоуровневой архитектуры с четким разделением ответственности между компонентами.
graph TB
subgraph "Frontend Layer"
A[React Frontend]
B[Mobile App]
end
subgraph "API Gateway"
C[FastAPI Backend]
D[Authentication]
E[Rate Limiting]
end
subgraph "Business Logic"
F[Document Management]
G[Risk Assessment]
H[PPE Management]
I[User Management]
end
subgraph "Data Layer"
J[PostgreSQL Database]
K[File Storage]
L[Cache Layer]
end
subgraph "External Services"
M[FNS API]
N[SMTP Server]
O[SFTP Storage]
end
A --> C
B --> C
C --> F
C --> G
C --> H
C --> I
F --> J
G --> J
H --> J
I --> J
F --> K
C --> L
C --> M
C --> N
C --> O🎯 Принципы архитектуры¶
1. Разделение ответственности (Separation of Concerns)¶
- Каждый компонент отвечает за свою область функциональности
- Четкие границы между слоями
- Независимость компонентов
2. Масштабируемость (Scalability)¶
- Горизонтальное масштабирование
- Асинхронная обработка
- Кэширование данных
3. Безопасность (Security)¶
- Многоуровневая аутентификация
- Авторизация на уровне ресурсов
- Шифрование данных
4. Надежность (Reliability)¶
- Обработка ошибок
- Резервное копирование
- Мониторинг системы
🖥️ Frontend Layer¶
React Frontend¶
- Технологии: React 18, TypeScript, Vite
- UI Framework: shadcn/ui + Tailwind CSS
- State Management: TanStack Query
- Routing: Wouter
// Пример структуры компонента
interface DocumentFormProps {
documentId: number;
onSubmit: (data: FormData) => void;
}
const DocumentForm: React.FC<DocumentFormProps> = ({ documentId, onSubmit }) => {
const { data: document } = useQuery(['document', documentId],
() => fetchDocument(documentId)
);
return (
<form onSubmit={handleSubmit}>
{/* Форма документа */}
</form>
);
};
Мобильное приложение¶
- Технологии: React Native
- Функциональность: просмотр документов, уведомления, заполнение форм
- Синхронизация: офлайн-режим с последующей синхронизацией
🔌 API Gateway¶
FastAPI Backend¶
- Технологии: Python 3.11+, FastAPI, Pydantic
- Аутентификация: JWT токены
- Документация: автоматическая генерация OpenAPI/Swagger
# Пример API эндпоинта
@router.post("/documents/upload", response_model=DocumentResponse)
async def upload_document(
file: UploadFile = File(...),
name: str = Form(...),
current_user: User = Depends(get_current_active_user),
db: Session = Depends(get_db)
):
"""Загрузка нового документа"""
# Логика загрузки документа
pass
Middleware¶
- CORS: настройка для фронтенда
- Rate Limiting: ограничение запросов
- Logging: логирование всех запросов
- Error Handling: централизованная обработка ошибок
🧠 Business Logic Layer¶
Модули системы¶
1. Document Management¶
class DocumentService:
def upload_document(self, file: UploadFile, metadata: dict) -> Document:
"""Загрузка и обработка документа"""
pass
def extract_template_fields(self, file_path: str) -> dict:
"""Извлечение полей шаблона"""
pass
def generate_filled_document(self, template_id: int, data: dict) -> str:
"""Генерация заполненного документа"""
pass
2. Risk Assessment¶
class RiskAssessmentService:
def calculate_risk_score(self, severity: int, probability: int,
coefficients: dict) -> RiskResult:
"""Расчет балла риска"""
pass
def get_risk_recommendations(self, risk_level: str) -> list:
"""Получение рекомендаций по снижению риска"""
pass
3. PPE Management¶
class PPEService:
def get_ppe_norms(self, position_id: int) -> list:
"""Получение норм выдачи СИЗ для должности"""
pass
def track_ppe_issuance(self, employee_id: int, ppe_items: list) -> None:
"""Учет выдачи СИЗ"""
pass
🗄️ Data Layer¶
PostgreSQL Database¶
- Версия: PostgreSQL 14+
- ORM: SQLAlchemy 2.0
- Миграции: Alembic
- Индексы: оптимизированы для частых запросов
-- Пример структуры таблицы
CREATE TABLE safety_users (
id SERIAL PRIMARY KEY,
username VARCHAR(50) UNIQUE NOT NULL,
email VARCHAR(100) UNIQUE NOT NULL,
full_name VARCHAR(200) NOT NULL,
password_hash VARCHAR(255) NOT NULL,
role VARCHAR(20) DEFAULT 'user',
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_users_username ON safety_users(username);
CREATE INDEX idx_users_email ON safety_users(email);
CREATE INDEX idx_users_role ON safety_users(role);
File Storage¶
- Локальное хранилище: для разработки и тестирования
- SFTP: для продакшн среды
- Структура: организована по типам документов и организациям
uploads/
├── documents/
│ ├── templates/
│ ├── filled/
│ └── public/
├── organizations/
│ ├── {org_id}/
│ │ ├── documents/
│ │ ├── forms/
│ │ └── reports/
└── temp/
Cache Layer¶
- Redis: для кэширования сессий и часто запрашиваемых данных
- TTL: настраиваемое время жизни кэша
- Invalidation: автоматическая инвалидация при изменениях
🔗 External Services¶
FNS API Integration¶
class FNSService:
def get_organization_data(self, inn: str) -> OrganizationData:
"""Получение данных организации из ФНС"""
pass
def validate_inn(self, inn: str) -> bool:
"""Валидация ИНН"""
pass
Email Service¶
class EmailService:
def send_notification(self, to: str, subject: str, body: str) -> bool:
"""Отправка email уведомления"""
pass
def send_bulk_notifications(self, recipients: list, message: dict) -> int:
"""Массовая отправка уведомлений"""
pass
🔒 Security Architecture¶
Аутентификация¶
sequenceDiagram
participant U as User
participant F as Frontend
participant A as API
participant D as Database
U->>F: Login credentials
F->>A: POST /auth/login
A->>D: Validate credentials
D-->>A: User data
A-->>F: JWT token
F-->>U: Success + tokenАвторизация¶
- RBAC: Role-Based Access Control
- Уровни доступа: system, organization, user
- Проверка прав: на каждом эндпоинте
def require_organization_membership(organization_id: int):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
current_user = kwargs.get('current_user')
if not check_membership(current_user.id, organization_id):
raise HTTPException(403, "Access denied")
return await func(*args, **kwargs)
return wrapper
return decorator
📊 Monitoring & Logging¶
Логирование¶
- Структурированные логи: JSON формат
- Уровни: DEBUG, INFO, WARNING, ERROR, CRITICAL
- Ротация: по размеру и времени
- Централизованный сбор: ELK Stack
Мониторинг¶
- Метрики: производительность, использование ресурсов
- Алерты: автоматические уведомления о проблемах
- Дашборды: визуализация состояния системы
🚀 Deployment Architecture¶
Контейнеризация¶
# Dockerfile для backend
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
Orchestration¶
- Docker Compose: для разработки
- Kubernetes: для продакшн
- Load Balancer: распределение нагрузки
- Auto-scaling: автоматическое масштабирование
🔄 Data Flow¶
Типичный запрос¶
sequenceDiagram
participant U as User
participant F as Frontend
participant A as API
participant S as Service
participant D as Database
participant C as Cache
U->>F: User action
F->>A: API request + JWT
A->>A: Validate token
A->>C: Check cache
alt Cache hit
C-->>A: Cached data
else Cache miss
A->>S: Business logic
S->>D: Database query
D-->>S: Data
S-->>A: Processed data
A->>C: Store in cache
end
A-->>F: Response
F-->>U: UI update📈 Performance Considerations¶
Оптимизация базы данных¶
- Индексы: на часто используемых полях
- Партиционирование: для больших таблиц
- Connection pooling: переиспользование соединений
- Query optimization: оптимизированные запросы
Кэширование¶
- Application cache: часто используемые данные
- Database cache: результаты запросов
- CDN: статические ресурсы
- Browser cache: клиентское кэширование
Асинхронность¶
- Background tasks: тяжелые операции
- Message queues: обработка событий
- WebSockets: real-time уведомления
- Streaming: большие файлы
Архитектура системы спроектирована с учетом требований масштабируемости, безопасности и производительности.