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.
63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
import uuid
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.audit import service as audit
|
|
from app.db.session import set_space_context
|
|
from app.errors import conflict, forbidden, not_found
|
|
from app.identity.models import User
|
|
from app.spaces.models import FinancialSpace, Membership
|
|
|
|
|
|
def create_space(db: Session, *, user: User, name: str) -> FinancialSpace:
|
|
space = FinancialSpace(name=name)
|
|
db.add(space)
|
|
db.flush()
|
|
set_space_context(db, space.id)
|
|
db.add(Membership(space_id=space.id, user_id=user.id, role="owner"))
|
|
from app.categorization.service import ensure_default_categories
|
|
from app.ledger.savings import ensure_savings_resources
|
|
|
|
ensure_savings_resources(db, space_id=space.id)
|
|
ensure_default_categories(db, space_id=space.id)
|
|
audit.record(db, event_type="space.created", space_id=space.id, user_id=user.id)
|
|
return space
|
|
|
|
|
|
def get_membership(db: Session, *, space_id: uuid.UUID, user_id: uuid.UUID) -> Membership:
|
|
membership = db.query(Membership).filter_by(space_id=space_id, user_id=user_id).first()
|
|
if membership is None:
|
|
raise not_found("SPACE_NOT_FOUND", "Financial space not found")
|
|
return membership
|
|
|
|
|
|
def get_space_for_member(
|
|
db: Session, *, space_id: uuid.UUID, user: User
|
|
) -> tuple[FinancialSpace, Membership]:
|
|
space = db.get(FinancialSpace, space_id)
|
|
if space is None:
|
|
raise not_found("SPACE_NOT_FOUND", "Financial space not found")
|
|
membership = get_membership(db, space_id=space_id, user_id=user.id)
|
|
if space.status == "blocked":
|
|
raise conflict(
|
|
"SPACE_BLOCKED",
|
|
"Financial space is blocked pending deletion",
|
|
)
|
|
return space, membership
|
|
|
|
|
|
def require_owner(membership: Membership) -> None:
|
|
if membership.role != "owner":
|
|
raise forbidden("SPACE_OWNER_REQUIRED", "Owner role required")
|
|
|
|
|
|
def list_spaces_for_user(db: Session, *, user: User) -> list[tuple[FinancialSpace, Membership]]:
|
|
rows = (
|
|
db.query(FinancialSpace, Membership)
|
|
.join(Membership, Membership.space_id == FinancialSpace.id)
|
|
.filter(Membership.user_id == user.id)
|
|
.order_by(FinancialSpace.created_at)
|
|
.all()
|
|
)
|
|
return [(space, membership) for space, membership in rows]
|