budget

No hooks declared

• TokenTypeCreated v1
  After a new token type is created
  payload: {token_type_id, owner_id, name, occurred_at}
• TokenTypeUpdated v1
  After a token type is modified
  payload: {token_type_id, owner_id, name, occurred_at}
• TokenTypeDeleted v1
  After a token type is deleted
  payload: {token_type_id, owner_id, occurred_at}
• BudgetPlanCreated v1
  After a new budget plan is created
  payload: {plan_id, owner_id, period_type, occurred_at}
• BudgetPlanUpdated v1
  After a budget plan is modified
  payload: {plan_id, owner_id, period_type, occurred_at}
• BudgetGranted v1
  When budget is granted for a period
  payload: {owner_id, period, token_type_id, amount, occurred_at}
• TaskScheduled v1
  After a task is scheduled with budget
  payload: {scheduled_task_id, task_id, owner_id, period, cost: {token_type_id: amount}, occurred_at}
• TaskUnscheduled v1
  After a task is unscheduled and budget released
  payload: {scheduled_task_id, task_id, owner_id, period, released_cost, occurred_at}
• TaskRescheduled v1
  After a task is moved to a different period
  payload: {scheduled_task_id, task_id, owner_id, old_period, new_period, cost, occurred_at}
• BudgetSpent v1
  When task budget is converted from reserved to spent
  payload: {scheduled_task_id, task_id, owner_id, period, cost, occurred_at}
• BudgetBorrowed v1
  When budget is borrowed to cover a shortfall
  payload: {debt_id, owner_id, period, token_type_id, amount, task_id, occurred_at}
• DebtRepaid v1
  After a debt obligation is repaid
  payload: {debt_id, owner_id, token_type_id, amount, remaining, occurred_at}
• DebtForgiven v1
  After a debt obligation is forgiven
  payload: {debt_id, owner_id, token_type_id, forgiven_amount, occurred_at}
• TaskSplit v1
  After a task's budget is split across periods
  payload: {task_id, owner_id, periods, cost_per_period, occurred_at}
• BudgetRolledOver v1
  When unused budget is rolled over to next period
  payload: {owner_id, from_period, to_period, token_type_id, amount, capped_from, occurred_at}
• BudgetAutoBalanced v1
  When tasks are auto-balanced within a period
  payload: {owner_id, period, adjustments: {task_id: {token_type_id: delta}}, occurred_at}
• TransactionRecorded v1
  After a ledger transaction is recorded
  payload: {transaction_id, owner_id, period, token_type_id, amount, transaction_type, task_id, occurred_at}

from uuid import uuid4
from datetime import date

from psp.components.budget.application.use_cases import ScheduleTask
from psp.components.budget.domain import Period, PeriodType, TaskCost
from psp.platform.clock import SystemClock
from psp.platform.eventbus import InMemoryEventBus

# Wire dependencies
schedule_task = ScheduleTask(
    scheduled_task_repo=scheduled_task_repo,
    budget_plan_repo=budget_plan_repo,
    ledger_repo=ledger_repo,
    clock=SystemClock(),
    event_bus=InMemoryEventBus(),
)

# Schedule task for today
result = schedule_task.execute(
    task_id=task_id,
    owner_id=current_user_id,
    period=Period.today(PeriodType.DAY),
    cost=TaskCost(costs={deep_work_token_id: 2}),
)

print(f"Scheduled: {result.scheduled_task_id}")

Use ScheduleTask use case to reserve budget for a task.

from psp.components.budget.application.use_cases import CheckBudget
from psp.components.budget.domain import Period, PeriodType, TaskCost

check = CheckBudget(
    budget_plan_repo=budget_plan_repo,
    scheduled_task_repo=scheduled_task_repo,
    ledger_repo=ledger_repo,
)

result = check.execute(
    owner_id=current_user_id,
    period=Period.today(PeriodType.DAY),
    cost=TaskCost(costs={deep_work_token_id: 3}),
)

if result.can_schedule:
    print("Budget available!")
else:
    print(f"Over budget by: {result.shortfall}")

Use CheckBudget to see if a task can be scheduled.

from psp.components.budget.application.use_cases import GetPeriodSnapshot
from psp.components.budget.domain import Period, PeriodType

get_snapshot = GetPeriodSnapshot(
    budget_plan_repo=budget_plan_repo,
    scheduled_task_repo=scheduled_task_repo,
    ledger_repo=ledger_repo,
)

snapshot = get_snapshot.execute(
    owner_id=current_user_id,
    period=Period.today(PeriodType.DAY),
)

for token_id, remaining in snapshot.remaining.items():
    print(f"Token {token_id}: {remaining} remaining")

View budget status for a period.