busya
1e6ec34c2b
- Removes source_dialect and target_dialect from TerminologyDictionary model, schemas, routes, helper serialization, and all tests - Adds allowed_languages config field (BCP-47 language codes) to GlobalSettings with validation, consolidated settings response, and API endpoint - Adds alembic migration f1a2b3c4d5e6 to drop the two columns
71 lines
2.6 KiB
Python
71 lines
2.6 KiB
Python
# #region TranslateHelpersModule [C:2] [TYPE Module] [SEMANTICS sqlalchemy, translate, helper, query, mapping]
|
|
# @BRIEF Shared helper functions for translate route handlers.
|
|
# @LAYER API
|
|
# @RELATION DEPENDS_ON -> [EXT:frontend:models.translate]
|
|
|
|
from typing import Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
# #region _run_to_response [C:2] [TYPE Function]
|
|
# @BRIEF Convert TranslationRun ORM to response dict.
|
|
def _run_to_response(run: Any) -> dict:
|
|
return {
|
|
"id": run.id,
|
|
"job_id": run.job_id,
|
|
"status": run.status,
|
|
"trigger_type": run.trigger_type,
|
|
"started_at": run.started_at.isoformat() if run.started_at else None,
|
|
"completed_at": run.completed_at.isoformat() if run.completed_at else None,
|
|
"error_message": run.error_message,
|
|
"total_records": run.total_records or 0,
|
|
"successful_records": run.successful_records or 0,
|
|
"failed_records": run.failed_records or 0,
|
|
"skipped_records": run.skipped_records or 0,
|
|
"cache_hits": run.cache_hits or 0,
|
|
"insert_status": run.insert_status,
|
|
"superset_execution_id": run.superset_execution_id,
|
|
"config_snapshot": run.config_snapshot,
|
|
"key_hash": run.key_hash,
|
|
"config_hash": run.config_hash,
|
|
"dict_snapshot_hash": run.dict_snapshot_hash,
|
|
"created_by": run.created_by,
|
|
"created_at": run.created_at.isoformat() if run.created_at else None,
|
|
}
|
|
# #endregion _run_to_response
|
|
|
|
|
|
# #region _dict_to_response [C:2] [TYPE Function]
|
|
# @BRIEF Convert a TerminologyDictionary ORM model to a response dict with computed entry_count.
|
|
def _dict_to_response(d: Any, entry_count: int = 0) -> dict:
|
|
return {
|
|
"id": d.id,
|
|
"name": d.name,
|
|
"description": d.description,
|
|
"is_active": d.is_active,
|
|
"created_by": d.created_by,
|
|
"created_at": d.created_at,
|
|
"updated_at": d.updated_at,
|
|
"entry_count": entry_count,
|
|
}
|
|
# #endregion _dict_to_response
|
|
|
|
|
|
# #region _get_dictionary_entry_counts [C:2] [TYPE Function]
|
|
# @BRIEF Get entry counts for a list of dictionary IDs.
|
|
def _get_dictionary_entry_counts(db: Session, dict_ids: list[str]) -> dict[str, int]:
|
|
from sqlalchemy import func
|
|
|
|
from ....models.translate import DictionaryEntry
|
|
counts = (
|
|
db.query(DictionaryEntry.dictionary_id, func.count(DictionaryEntry.id))
|
|
.filter(DictionaryEntry.dictionary_id.in_(dict_ids))
|
|
.group_by(DictionaryEntry.dictionary_id)
|
|
.all()
|
|
)
|
|
return {row[0]: row[1] for row in counts}
|
|
# #endregion _get_dictionary_entry_counts
|
|
|
|
# #endregion TranslateHelpersModule
|