latest version of apis event and cahce ablitites added

This commit is contained in:
2025-02-10 11:41:38 +03:00
parent e832ec7603
commit 26f601f01a
396 changed files with 34981 additions and 2 deletions

View File

@@ -0,0 +1,9 @@
# Original content from ApiEvents/ValidationServiceApi/schema_converter.py
from typing import Dict, Any, Type, get_type_hints, get_args, get_origin
from pydantic import BaseModel, Field, EmailStr
from enum import Enum
import inspect
from fastapi import APIRouter
from datetime import datetime
# ... rest of the file content ...

View File

@@ -0,0 +1,149 @@
from typing import Dict, Any, Type, Optional
from pydantic import BaseModel
from fastapi import APIRouter, Header
class ValidationMessages(BaseModel):
"""Messages for Zod validation"""
required: str
invalid_type: str
invalid_string: Dict[str, str] # email, url, etc
too_small: Dict[str, str] # string, array, number
too_big: Dict[str, str] # string, array, number
invalid_date: str
invalid_enum: str
custom: Dict[str, str]
class SchemaField(BaseModel):
"""Schema field definition"""
type: str
items: Optional[str] = None # For arrays
values: Optional[list] = None # For enums
validations: Optional[Dict[str, Any]] = None
class SchemaDefinition(BaseModel):
"""Complete schema definition"""
name: str
fields: Dict[str, SchemaField]
messages: ValidationMessages
class UnifiedSchemaService:
def __init__(self):
self.messages = {
"tr": ValidationMessages(
required="Bu alan zorunludur",
invalid_type="Geçersiz tip",
invalid_string={
"email": "Geçerli bir e-posta adresi giriniz",
"url": "Geçerli bir URL giriniz",
"uuid": "Geçerli bir UUID giriniz",
},
too_small={
"string": "{min} karakterden az olamaz",
"array": "En az {min} öğe gereklidir",
"number": "En az {min} olmalıdır",
},
too_big={
"string": "{max} karakterden fazla olamaz",
"array": "En fazla {max} öğe olabilir",
"number": "En fazla {max} olabilir",
},
invalid_date="Geçerli bir tarih giriniz",
invalid_enum="Geçersiz seçim",
custom={
"password_match": "Şifreler eşleşmiyor",
"strong_password": "Şifre güçlü değil",
},
),
"en": ValidationMessages(
required="This field is required",
invalid_type="Invalid type",
invalid_string={
"email": "Please enter a valid email",
"url": "Please enter a valid URL",
"uuid": "Please enter a valid UUID",
},
too_small={
"string": "Must be at least {min} characters",
"array": "Must contain at least {min} items",
"number": "Must be at least {min}",
},
too_big={
"string": "Must be at most {max} characters",
"array": "Must contain at most {max} items",
"number": "Must be at most {max}",
},
invalid_date="Please enter a valid date",
invalid_enum="Invalid selection",
custom={
"password_match": "Passwords do not match",
"strong_password": "Password is not strong enough",
},
),
}
def get_schema_with_messages(
self, model: Type[BaseModel], lang: str = "tr"
) -> SchemaDefinition:
"""Get schema definition with validation messages"""
fields: Dict[str, SchemaField] = {}
for field_name, field in model.__fields__.items():
field_info = SchemaField(
type=self._get_field_type(field.outer_type_),
items=self._get_items_type(field.outer_type_),
values=self._get_enum_values(field.outer_type_),
validations=self._get_validations(field),
)
fields[field_name] = field_info
return SchemaDefinition(
name=model.__name__, fields=fields, messages=self.messages[lang]
)
def _get_field_type(self, type_: Type) -> str:
# Implementation similar to SchemaConverter
pass
def _get_items_type(self, type_: Type) -> Optional[str]:
# Implementation similar to SchemaConverter
pass
def _get_enum_values(self, type_: Type) -> Optional[list]:
# Implementation similar to SchemaConverter
pass
def _get_validations(self, field) -> Optional[Dict[str, Any]]:
# Implementation similar to SchemaConverter
pass
router = APIRouter(prefix="/api/schema", tags=["Schema"])
schema_service = UnifiedSchemaService()
@router.get("/model/{model_name}")
async def get_model_schema(
model_name: str, accept_language: Optional[str] = Header(default="tr")
) -> SchemaDefinition:
"""Get model schema with validation messages"""
# You'd need to implement model lookup
models = {
"User": UserModel,
"Product": ProductModel,
# Add your models here
}
if model_name not in models:
raise ValueError(f"Model {model_name} not found")
lang = accept_language.split(",")[0][:2]
return schema_service.get_schema_with_messages(
models[model_name], lang if lang in ["tr", "en"] else "tr"
)