131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
# [DEF:backend.src.api.routes.mappings:Module]
|
|
#
|
|
# @TIER: STANDARD
|
|
# @SEMANTICS: api, mappings, database, fuzzy-matching
|
|
# @PURPOSE: API endpoints for managing database mappings and getting suggestions.
|
|
# @LAYER: API
|
|
# @RELATION: DEPENDS_ON -> backend.src.dependencies
|
|
# @RELATION: DEPENDS_ON -> backend.src.core.database
|
|
# @RELATION: DEPENDS_ON -> backend.src.services.mapping_service
|
|
#
|
|
# @INVARIANT: Mappings are persisted in the SQLite database.
|
|
|
|
# [SECTION: IMPORTS]
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from typing import List, Optional
|
|
from ...core.logger import belief_scope
|
|
from ...dependencies import get_config_manager, has_permission
|
|
from ...core.database import get_db
|
|
from ...models.mapping import DatabaseMapping
|
|
from pydantic import BaseModel
|
|
# [/SECTION]
|
|
|
|
router = APIRouter(tags=["mappings"])
|
|
|
|
# [DEF:MappingCreate:DataClass]
|
|
class MappingCreate(BaseModel):
|
|
source_env_id: str
|
|
target_env_id: str
|
|
source_db_uuid: str
|
|
target_db_uuid: str
|
|
source_db_name: str
|
|
target_db_name: str
|
|
engine: Optional[str] = None
|
|
# [/DEF:MappingCreate:DataClass]
|
|
|
|
# [DEF:MappingResponse:DataClass]
|
|
class MappingResponse(BaseModel):
|
|
id: str
|
|
source_env_id: str
|
|
target_env_id: str
|
|
source_db_uuid: str
|
|
target_db_uuid: str
|
|
source_db_name: str
|
|
target_db_name: str
|
|
engine: Optional[str] = None
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
# [/DEF:MappingResponse:DataClass]
|
|
|
|
# [DEF:SuggestRequest:DataClass]
|
|
class SuggestRequest(BaseModel):
|
|
source_env_id: str
|
|
target_env_id: str
|
|
# [/DEF:SuggestRequest:DataClass]
|
|
|
|
# [DEF:get_mappings:Function]
|
|
# @PURPOSE: List all saved database mappings.
|
|
# @PRE: db session is injected.
|
|
# @POST: Returns filtered list of DatabaseMapping records.
|
|
@router.get("", response_model=List[MappingResponse])
|
|
async def get_mappings(
|
|
source_env_id: Optional[str] = None,
|
|
target_env_id: Optional[str] = None,
|
|
db: Session = Depends(get_db),
|
|
_ = Depends(has_permission("plugin:mapper", "EXECUTE"))
|
|
):
|
|
with belief_scope("get_mappings"):
|
|
query = db.query(DatabaseMapping)
|
|
if source_env_id:
|
|
query = query.filter(DatabaseMapping.source_env_id == source_env_id)
|
|
if target_env_id:
|
|
query = query.filter(DatabaseMapping.target_env_id == target_env_id)
|
|
return query.all()
|
|
# [/DEF:get_mappings:Function]
|
|
|
|
# [DEF:create_mapping:Function]
|
|
# @PURPOSE: Create or update a database mapping.
|
|
# @PRE: mapping is valid MappingCreate, db session is injected.
|
|
# @POST: DatabaseMapping created or updated in database.
|
|
@router.post("", response_model=MappingResponse)
|
|
async def create_mapping(
|
|
mapping: MappingCreate,
|
|
db: Session = Depends(get_db),
|
|
_ = Depends(has_permission("plugin:mapper", "EXECUTE"))
|
|
):
|
|
with belief_scope("create_mapping"):
|
|
# Check if mapping already exists
|
|
existing = db.query(DatabaseMapping).filter(
|
|
DatabaseMapping.source_env_id == mapping.source_env_id,
|
|
DatabaseMapping.target_env_id == mapping.target_env_id,
|
|
DatabaseMapping.source_db_uuid == mapping.source_db_uuid
|
|
).first()
|
|
|
|
if existing:
|
|
existing.target_db_uuid = mapping.target_db_uuid
|
|
existing.target_db_name = mapping.target_db_name
|
|
existing.engine = mapping.engine
|
|
db.commit()
|
|
db.refresh(existing)
|
|
return existing
|
|
|
|
new_mapping = DatabaseMapping(**mapping.dict())
|
|
db.add(new_mapping)
|
|
db.commit()
|
|
db.refresh(new_mapping)
|
|
return new_mapping
|
|
# [/DEF:create_mapping:Function]
|
|
|
|
# [DEF:suggest_mappings_api:Function]
|
|
# @PURPOSE: Get suggested mappings based on fuzzy matching.
|
|
# @PRE: request is valid SuggestRequest, config_manager is injected.
|
|
# @POST: Returns mapping suggestions.
|
|
@router.post("/suggest")
|
|
async def suggest_mappings_api(
|
|
request: SuggestRequest,
|
|
config_manager=Depends(get_config_manager),
|
|
_ = Depends(has_permission("plugin:mapper", "EXECUTE"))
|
|
):
|
|
with belief_scope("suggest_mappings_api"):
|
|
from ...services.mapping_service import MappingService
|
|
service = MappingService(config_manager)
|
|
try:
|
|
return await service.get_suggestions(request.source_env_id, request.target_env_id)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
# [/DEF:suggest_mappings_api:Function]
|
|
|
|
# [/DEF:backend.src.api.routes.mappings:Module]
|