k1nq
63d065011b
- Updated timestamps and AST hashes in manifest.json for various backend and frontend files to reflect recent changes. - Added new files related to MCC categories and optional hold reasons in the manifest. - Modified the README in the infra directory to correct the linting tool from 'mypy' to 'ty' for type checking in the backend project.
379 lines
12 KiB
Python
379 lines
12 KiB
Python
import uuid
|
|
from collections import defaultdict
|
|
from datetime import date
|
|
from typing import Annotated, Literal
|
|
|
|
from fastapi import APIRouter, Query
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.budgeting import service
|
|
from app.categorization import service as categorization
|
|
from app.categorization.models import CategoryMcc
|
|
from app.dependencies import Context, Csrf, Db
|
|
|
|
router = APIRouter(prefix="/budget", tags=["budget"])
|
|
|
|
|
|
class CategoryBody(BaseModel):
|
|
name: str = Field(min_length=1, max_length=200)
|
|
kind: Literal["monthly", "savings_envelope"] = "monthly"
|
|
mcc_codes: list[int] = Field(default_factory=list, alias="mccCodes")
|
|
|
|
|
|
class AssignmentItem(BaseModel):
|
|
category_id: uuid.UUID
|
|
planned_kopecks: int = Field(ge=0)
|
|
|
|
|
|
class AssignmentsBody(BaseModel):
|
|
items: list[AssignmentItem]
|
|
|
|
|
|
class FundingBody(BaseModel):
|
|
category_id: uuid.UUID
|
|
amount_kopecks: int = Field(gt=0)
|
|
kind: Literal["fund", "unfund"] = "fund"
|
|
note: str | None = Field(default=None, max_length=500)
|
|
|
|
|
|
class ExpectedIncomeBody(BaseModel):
|
|
expected_on: date
|
|
amount_kopecks: int = Field(gt=0)
|
|
source_name: str | None = Field(default=None, max_length=200)
|
|
note: str | None = Field(default=None, max_length=500)
|
|
|
|
|
|
def period_for(db: Db, context: Context, year: int, month: int):
|
|
return service.get_or_create_period(db, space_id=context.space.id, year=year, month=month)
|
|
|
|
|
|
class MonthBody(BaseModel):
|
|
month: str
|
|
|
|
|
|
class PlanBody(MonthBody):
|
|
plannedKopecks: int = Field(ge=0)
|
|
|
|
|
|
class CategoryFundingBody(MonthBody):
|
|
amountKopecks: int = Field(gt=0)
|
|
|
|
|
|
def parse_month(value: str) -> tuple[int, int]:
|
|
try:
|
|
year_text, month_text = value.split("-", 1)
|
|
year, month = int(year_text), int(month_text)
|
|
except (ValueError, AttributeError) as exc:
|
|
from app.errors import bad_request
|
|
|
|
raise bad_request("BUDGET_PERIOD_INVALID", "Month must use YYYY-MM") from exc
|
|
service.validate_year_month(year, month)
|
|
return year, month
|
|
|
|
|
|
def frontend_summary(state: dict) -> dict:
|
|
monthly = state["monthly_categories"]
|
|
envelopes = state["envelopes"]
|
|
funded = sum(row["funded"] for row in monthly) + sum(row["funded_total"] for row in envelopes)
|
|
spent = sum(row["spent"] for row in monthly)
|
|
available = sum(row["available"] for row in monthly + envelopes)
|
|
return {
|
|
"month": f"{state['period']['year']:04d}-{state['period']['month']:02d}",
|
|
"status": state["period"]["status"],
|
|
"projectedPoolKopecks": state["carry_pool"]
|
|
+ state["expected_income_total"]
|
|
- state["planned_total"],
|
|
"actualPoolKopecks": state["pool"],
|
|
"plannedTotalKopecks": state["planned_total"],
|
|
"fundedTotalKopecks": funded,
|
|
"spentTotalKopecks": spent,
|
|
"availableKopecks": available,
|
|
"warnings": [{"code": code} for code in state["warnings"]],
|
|
"blockers": state["close_blockers"],
|
|
"expectedIncomes": [
|
|
{
|
|
"id": item["id"],
|
|
"expectedOn": item["expected_on"],
|
|
"amountKopecks": item["amount_kopecks"],
|
|
"sourceName": item["source_name"],
|
|
"matchedKopecks": item["matched_kopecks"],
|
|
"status": "matched"
|
|
if item["matched_kopecks"] >= item["amount_kopecks"]
|
|
else "partially_matched"
|
|
if item["matched_kopecks"]
|
|
else "expected",
|
|
}
|
|
for item in state["expected_incomes"]
|
|
],
|
|
}
|
|
|
|
|
|
@router.get("/summary")
|
|
def summary(month: str, db: Db, context: Context) -> dict:
|
|
year, month_number = parse_month(month)
|
|
period = period_for(db, context, year, month_number)
|
|
state = service.compute_period_state(db, space_id=context.space.id, period=period)
|
|
return frontend_summary(state)
|
|
|
|
|
|
@router.get("/categories")
|
|
def categories(month: str, db: Db, context: Context) -> list[dict]:
|
|
year, month_number = parse_month(month)
|
|
period = period_for(db, context, year, month_number)
|
|
state = service.compute_period_state(db, space_id=context.space.id, period=period)
|
|
rows = state["monthly_categories"] + state["envelopes"]
|
|
category_by_id = {
|
|
str(item.id): item
|
|
for item in service.list_categories(db, space_id=context.space.id, include_archived=True)
|
|
}
|
|
result = []
|
|
for row in rows:
|
|
category = category_by_id[row["category_id"]]
|
|
result.append(
|
|
{
|
|
"id": row["category_id"],
|
|
"name": row["name"],
|
|
"nameKey": row["name_key"],
|
|
"kind": category.kind,
|
|
"system": row["system"],
|
|
"archived": category.archived,
|
|
"creditAccountId": category.linked_account_id,
|
|
"plannedKopecks": row["planned"],
|
|
"fundedKopecks": row["funded"],
|
|
"spentKopecks": row["spent"],
|
|
"availableKopecks": row["available"],
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def category_definition_json(category, mcc_codes: list[int]) -> dict:
|
|
return {
|
|
"id": category.id,
|
|
"name": categorization.resolve_category_name(category),
|
|
"nameKey": categorization.resolve_category_name_key(category),
|
|
"kind": category.kind,
|
|
"systemKey": category.system_key,
|
|
"linkedAccountId": category.linked_account_id,
|
|
"archived": category.archived,
|
|
"mccCodes": mcc_codes,
|
|
}
|
|
|
|
|
|
@router.get("/category-definitions")
|
|
def category_definitions(
|
|
db: Db,
|
|
context: Context,
|
|
include_archived: Annotated[bool, Query(alias="includeArchived")] = False,
|
|
) -> list[dict]:
|
|
items = service.list_categories(
|
|
db, space_id=context.space.id, include_archived=include_archived
|
|
)
|
|
codes_by_category: dict[uuid.UUID, list[int]] = defaultdict(list)
|
|
for category_id, code in (
|
|
db.query(CategoryMcc.category_id, CategoryMcc.mcc_code)
|
|
.filter_by(space_id=context.space.id)
|
|
.order_by(CategoryMcc.mcc_code)
|
|
):
|
|
codes_by_category[category_id].append(code)
|
|
return [category_definition_json(item, codes_by_category[item.id]) for item in items]
|
|
|
|
|
|
class CategoryDefinitionPatchBody(BaseModel):
|
|
name: str | None = Field(default=None, min_length=1, max_length=200)
|
|
archived: bool | None = None
|
|
mcc_codes: list[int] | None = Field(default=None, alias="mccCodes")
|
|
|
|
|
|
@router.patch("/category-definitions/{category_id}")
|
|
def patch_category_definition(
|
|
category_id: uuid.UUID,
|
|
body: CategoryDefinitionPatchBody,
|
|
db: Db,
|
|
context: Context,
|
|
_csrf: Csrf,
|
|
) -> dict:
|
|
category = service.get_category(db, space_id=context.space.id, category_id=category_id)
|
|
if body.name is not None or body.archived is not None:
|
|
category = service.update_category_definition(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
category_id=category_id,
|
|
name=body.name,
|
|
archived=body.archived,
|
|
)
|
|
if body.mcc_codes is not None:
|
|
categorization.set_category_mcc_codes(
|
|
db,
|
|
space_id=context.space.id,
|
|
category_id=category_id,
|
|
codes=body.mcc_codes,
|
|
)
|
|
return category_definition_json(
|
|
category,
|
|
categorization.get_category_mcc_codes(
|
|
db, space_id=context.space.id, category_id=category.id
|
|
),
|
|
)
|
|
|
|
|
|
@router.post("/categories", status_code=201)
|
|
def create_category(body: CategoryBody, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
category = service.create_category(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
kind=body.kind,
|
|
name=body.name,
|
|
)
|
|
if body.mcc_codes:
|
|
categorization.set_category_mcc_codes(
|
|
db,
|
|
space_id=context.space.id,
|
|
category_id=category.id,
|
|
codes=body.mcc_codes,
|
|
)
|
|
return category_definition_json(
|
|
category,
|
|
categorization.get_category_mcc_codes(
|
|
db, space_id=context.space.id, category_id=category.id
|
|
),
|
|
)
|
|
|
|
|
|
@router.patch("/categories/{category_id}")
|
|
def update_plan(
|
|
category_id: uuid.UUID, body: PlanBody, db: Db, context: Context, _csrf: Csrf
|
|
) -> dict:
|
|
year, month_number = parse_month(body.month)
|
|
period = period_for(db, context, year, month_number)
|
|
service.set_assignments(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
period=period,
|
|
items=[{"category_id": category_id, "planned_kopecks": body.plannedKopecks}],
|
|
)
|
|
return next(item for item in categories(body.month, db, context) if item["id"] == category_id)
|
|
|
|
|
|
@router.post("/categories/{category_id}/fund")
|
|
def fund_category(
|
|
category_id: uuid.UUID,
|
|
body: CategoryFundingBody,
|
|
db: Db,
|
|
context: Context,
|
|
_csrf: Csrf,
|
|
) -> dict:
|
|
year, month_number = parse_month(body.month)
|
|
period = period_for(db, context, year, month_number)
|
|
service.fund(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
period=period,
|
|
category_id=category_id,
|
|
amount_kopecks=body.amountKopecks,
|
|
kind="fund",
|
|
)
|
|
return next(item for item in categories(body.month, db, context) if item["id"] == category_id)
|
|
|
|
|
|
@router.put("/{year}/{month}/assignments")
|
|
def assignments(
|
|
year: int, month: int, body: AssignmentsBody, db: Db, context: Context, _csrf: Csrf
|
|
) -> dict:
|
|
period = period_for(db, context, year, month)
|
|
rows = service.set_assignments(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
period=period,
|
|
items=[item.model_dump() for item in body.items],
|
|
)
|
|
return {"count": len(rows)}
|
|
|
|
|
|
@router.post("/{year}/{month}/funding", status_code=201)
|
|
def fund(year: int, month: int, body: FundingBody, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
period = period_for(db, context, year, month)
|
|
movement = service.fund(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
period=period,
|
|
**body.model_dump(),
|
|
)
|
|
return {"id": movement.id}
|
|
|
|
|
|
@router.post("/{year}/{month}/expected-incomes", status_code=201)
|
|
def expected_income(
|
|
year: int,
|
|
month: int,
|
|
body: ExpectedIncomeBody,
|
|
db: Db,
|
|
context: Context,
|
|
_csrf: Csrf,
|
|
) -> dict:
|
|
period = period_for(db, context, year, month)
|
|
income = service.add_expected_income(
|
|
db,
|
|
space_id=context.space.id,
|
|
user=context.user,
|
|
period=period,
|
|
**body.model_dump(),
|
|
)
|
|
return {"id": income.id}
|
|
|
|
|
|
@router.post("/{year}/{month}/close")
|
|
def close(year: int, month: int, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
period = period_for(db, context, year, month)
|
|
return service.close_period(db, space_id=context.space.id, user=context.user, period=period)
|
|
|
|
|
|
@router.post("/{year}/{month}/reopen")
|
|
def reopen(year: int, month: int, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
period = period_for(db, context, year, month)
|
|
affected = service.reopen_period(
|
|
db, space_id=context.space.id, user=context.user, period=period
|
|
)
|
|
return {"affected_period_ids": [item.id for item in affected]}
|
|
|
|
|
|
@router.post("/{year}/{month}/copy-previous")
|
|
def copy_previous(year: int, month: int, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
period = period_for(db, context, year, month)
|
|
return service.copy_previous_month(
|
|
db, space_id=context.space.id, user=context.user, period=period
|
|
)
|
|
|
|
|
|
@router.post("/close")
|
|
def close_frontend(body: MonthBody, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
year, month_number = parse_month(body.month)
|
|
period = period_for(db, context, year, month_number)
|
|
service.close_period(db, space_id=context.space.id, user=context.user, period=period)
|
|
state = service.compute_period_state(db, space_id=context.space.id, period=period)
|
|
return frontend_summary(state)
|
|
|
|
|
|
@router.post("/reopen")
|
|
def reopen_frontend(body: MonthBody, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
year, month_number = parse_month(body.month)
|
|
period = period_for(db, context, year, month_number)
|
|
service.reopen_period(db, space_id=context.space.id, user=context.user, period=period)
|
|
state = service.compute_period_state(db, space_id=context.space.id, period=period)
|
|
return frontend_summary(state)
|
|
|
|
|
|
@router.post("/copy-previous")
|
|
def copy_frontend(body: MonthBody, db: Db, context: Context, _csrf: Csrf) -> dict:
|
|
year, month_number = parse_month(body.month)
|
|
period = period_for(db, context, year, month_number)
|
|
service.copy_previous_month(db, space_id=context.space.id, user=context.user, period=period)
|
|
state = service.compute_period_state(db, space_id=context.space.id, period=period)
|
|
return frontend_summary(state)
|