from __future__ import annotations from typing import Any from fastapi import Request from fastapi.responses import RedirectResponse from sqlalchemy import select from app.modules.core.tenancy.models import FinancialYear def active_financial_year_from_request(request: Request) -> str | None: session = request.scope.get("session") or {} value = session.get("active_financial_year") or getattr(request.state, "year_code", None) value = (value or "").strip() if not value or value.upper() == "ALL": return None return value def get_financial_year(db: Any, *, tenant_id: int, year_code: str | None) -> FinancialYear | None: code = (year_code or "").strip() if not tenant_id or not code: return None return db.execute( select(FinancialYear).where( FinancialYear.tenant_id == int(tenant_id), FinancialYear.year_code == code, ) ).scalar_one_or_none() def is_financial_year_locked(db: Any, *, tenant_id: int, year_code: str | None) -> bool: fy = get_financial_year(db, tenant_id=tenant_id, year_code=year_code) return bool(fy and fy.is_locked) def redirect_if_financial_year_locked( db: Any, *, tenant_id: int, year_code: str | None, redirect_url: str, ) -> RedirectResponse | None: if is_financial_year_locked(db, tenant_id=tenant_id, year_code=year_code): separator = "&" if "?" in redirect_url else "?" return RedirectResponse(url=f"{redirect_url}{separator}year_locked=1", status_code=303) return None def is_row_financial_year_locked(db: Any, row: Any) -> bool: tenant_id = getattr(row, "tenant_id", None) year_code = getattr(row, "financial_year", None) if not tenant_id or not year_code: return False return is_financial_year_locked(db, tenant_id=int(tenant_id), year_code=str(year_code))