ducto.interface.memory module

In-memory credit store for testing and development.

class ducto.interface.memory.MemoryStore

Bases: CreditStore

Credit store backed by in-memory dicts. Zero dependencies.

Useful for unit testing and local development without a database. All data is lost when the process exits.

activate_pricing(version: int) → str

Activate a specific pricing version (deactivates all others).

Args: : version: The version number to activate.

Returns: : The activated config id.

add_credits(user_id: str, amount: int, type: str = 'adjustment', metadata: CreditMetadata | None = None, expires_at: datetime | None = None) → AddCreditsResult

Atomically add credits and log a transaction.

Args: : expires_at: Optional datetime after which the credits expire.

add_team_member(team_id: str, user_id: str, role: str = 'member', spend_cap: int | None = None) → AddTeamMemberResult

Add a user to a team.

Args: : team_id: The team’s UUID. user_id: The user’s UUID. role: Member role (e.g. “member”, “admin”). spend_cap: Optional per-user spend cap.

Returns: : AddTeamMemberResult confirming membership.

aggregate_stats(start: datetime, end: datetime) → AggregateStatsRow

Aggregate statistics across all users in a time window.

check_allowance(user_id: str) → AllowanceResult

Get remaining free allowance for current billing period.

check_spend_cap(user_id: str, model: str | None = None, amount: int | None = None) → CapCheckResult

Check whether a pending deduction would exceed any configured cap.

Args: : user_id: The user to check caps for. model: Optional model name for per-model caps. amount: The pending deduction amount.

Returns: : CapCheckResult with the check result.

create_team(name: str, initial_balance: int = 0) → CreateTeamResult

Create a team with a shared credit balance pool.

Args: : name: Human-readable team name. initial_balance: Starting credit balance.

Returns: : CreateTeamResult with the new team id.

daily_spend(start: datetime, end: datetime) → list[DailySpendRow]

Daily spend aggregation in a time window.

deduct_credits(user_id: str, reservation_id: str, amount: int, idempotency_key: str | None = None, metadata: CreditMetadata | None = None) → DeductionResult

Finalize a credit deduction and release the reservation.

If idempotency_key is provided and a matching transaction already exists, returns the existing result (idempotent replay).

deduct_team(team_id: str, user_id: str, amount: int, metadata: CreditMetadata | None = None) → TeamDeductionResult

Deduct credits from a team pool, attributed to a user.

Args: : team_id: The team’s UUID. user_id: The user to attribute the deduction to. amount: Credits to deduct. metadata: Extra metadata.

Returns: : TeamDeductionResult with transaction details.

get_active_pricing() → PricingConfigResult | None

Fetch the active pricing configuration from the store.

get_balance(user_id: str) → BalanceResult

Return current balance and lifetime purchased amount.

get_pricing_config(version: int) → PricingConfigResult | None

Fetch a specific pricing config by version number.

get_pricing_history() → list[PricingConfigHistoryItem]

List all pricing config versions (newest first).

get_team_balance(team_id: str) → TeamBalanceResult

Fetch team balance and member count.

Args: : team_id: The team’s UUID.

Returns: : TeamBalanceResult with balance and member count.

get_team_members(team_id: str) → list[TeamMember]

List all members of a team.

Args: : team_id: The team’s UUID.

Returns: : List of TeamMember.

get_user_plan(user_id: str) → GetUserPlanResult

Fetch user’s current plan (including feature entitlements).

increment_usage_window(user_id: str, plan_id: str, amount: int) → None

Record allowance consumption for current billing period.

refund_credits(transaction_id: str, amount: int | None = None, reason: str | None = None, metadata: CreditMetadata | None = None) → RefundResult

Refund a previous credit deduction.

Args: : transaction_id: The transaction to refund. amount: Optional partial refund amount. Full refund if omitted. reason: Optional reason for the refund. metadata: Extra metadata to attach to the refund transaction.

Returns: : RefundResult with the refund transaction details, or error set if the transaction doesn’t exist or is already refunded.

reserve_credits(user_id: str, amount: int, operation_type: str, metadata: CreditMetadata | None = None, min_balance: int = 5) → ReserveResult

Reserve credits for an upcoming operation.

Locks the user row to prevent concurrent overspend. Returns a ReserveResult with error set on failure.

set_active_pricing(config: PricingConfigData, label: str | None = None) → str

Publish a new pricing configuration.

Deactivates the previous active config and inserts a new one. Returns the new config id.

set_spend_cap(cap: SpendCap) → None

Configure a spend cap (MemoryStore-only helper for testing).

set_user_plan(user_id: str, plan_id: str) → SetUserPlanResult

Assign a plan to a user.

setup(database_url: str | None = None) → SetupResult

Run bundled SQL migrations (tables, indexes, RPCs).

Idempotent — safe to call on every deploy.

Args: : database_url: Postgres connection string. Required for stores : that manage schema setup directly (HttpxSupabaseStore, PostgresStore). Ignored by in-memory stores.

spend_by_model(start: datetime, end: datetime) → list[SpendByModelRow]

Aggregate spend by model in a time window.

spend_by_user(start: datetime, end: datetime) → list[SpendByUserRow]

Aggregate spend by user in a time window.

sweep_expired_credits(dry_run: bool = False) → SweepResult

Sweep expired credits from all users’ balances.

top_users(limit: int, start: datetime, end: datetime) → list[TopUserRow]

Top users by spend in a time window.