60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime, timedelta
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.errors import AppError
|
|
from app.models.data_quality import DataQualityIssue
|
|
|
|
|
|
def issue_due_at(detected_at: datetime, severity: str) -> datetime:
|
|
"""Return the local operational SLA deadline for a newly detected issue."""
|
|
return detected_at + {
|
|
"high": timedelta(hours=4),
|
|
"medium": timedelta(days=1),
|
|
"low": timedelta(days=3),
|
|
}.get(severity, timedelta(days=1))
|
|
|
|
|
|
def has_open_issue(
|
|
db: Session,
|
|
rule_type: str,
|
|
entity_type: str,
|
|
entity_id: uuid.UUID,
|
|
) -> bool:
|
|
return (
|
|
db.scalar(
|
|
select(DataQualityIssue.id).where(
|
|
DataQualityIssue.rule_type == rule_type,
|
|
DataQualityIssue.entity_type == entity_type,
|
|
DataQualityIssue.entity_id == entity_id,
|
|
DataQualityIssue.status == "open",
|
|
)
|
|
)
|
|
is not None
|
|
)
|
|
|
|
|
|
def new_scan_ref(prefix: str) -> str:
|
|
"""Generate a stable human-readable prefix with a concurrent-safe suffix."""
|
|
return f"{prefix}-{uuid.uuid4().hex[:10].upper()}"
|
|
|
|
|
|
def load_open_issue(db: Session, public_ref: str, *, lock: bool = True) -> DataQualityIssue:
|
|
statement = select(DataQualityIssue).where(DataQualityIssue.public_ref == public_ref)
|
|
if lock:
|
|
statement = statement.with_for_update().execution_options(populate_existing=True)
|
|
issue = db.scalar(statement)
|
|
if issue is None:
|
|
raise AppError("ISSUE_NOT_FOUND", "Data quality issue not found.", status_code=404)
|
|
if issue.status != "open":
|
|
raise AppError(
|
|
"ISSUE_NOT_OPEN",
|
|
f"Issue is '{issue.status}', not 'open'.",
|
|
status_code=409,
|
|
)
|
|
return issue
|