ducto.interface.models module

Pydantic schemas for credit store operations.

All store methods accept and return typed Pydantic models rather than raw dicts — validation at the boundary, clarity in the call sites.

class ducto.interface.models.AddCreditsResult(, transaction_id: str, user_id: str, amount: int, new_balance: int, lifetime_purchased: int = 0)

Bases: BaseModel

Result of adding credits to a user’s account.

amount : int

lifetime_purchased : int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

new_balance : int

transaction_id : str

user_id : str

class ducto.interface.models.AddTeamMemberResult(, team_id: str = '', user_id: str = '', role: str = 'member')

Bases: BaseModel

Result of adding a team member.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

role : str

team_id : str

user_id : str

class ducto.interface.models.AggregateStatsRow(, total_credits_consumed: int = 0, active_users: int = 0, avg_daily_spend: int = 0, top_model: str = '', top_user: str = '')

Bases: BaseModel

Aggregate statistics across all users in a time window.

active_users : int

avg_daily_spend : int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

top_model : str

top_user : str

total_credits_consumed : int

class ducto.interface.models.AllowanceResult(, plan_id: str, allowance_remaining: int, period_start: str, period_end: str)

Bases: BaseModel

Result of checking plan allowance.

allowance_remaining : int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

period_end : str

period_start : str

plan_id : str

class ducto.interface.models.BalanceResult(, user_id: str, balance: int = 0, lifetime_purchased: int = 0)

Bases: BaseModel

Current credit balance for a user.

balance : int

lifetime_purchased : int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

user_id : str

class ducto.interface.models.CapCheckResult(, capped: bool = False, current_spend: int = 0, cap_limit: int = 0, action: str | None = None, model: str | None = None)

Bases: BaseModel

Result of checking a spend cap.

action : str | None

cap_limit : int

capped : bool

current_spend : int

model : str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ducto.interface.models.CheckFeatureResult(, user_id: str, feature: str, value: Any = None, has_feature: bool = False)

Bases: BaseModel

Result of checking a user’s feature entitlement.

feature : str

has_feature : bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

user_id : str

value : Any

class ducto.interface.models.CreateTeamResult(, team_id: str = '', name: str = '')

Bases: BaseModel

Result of creating a team.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name : str

team_id : str

class ducto.interface.models.CreditMetadata(, input_tokens: int | None = None, output_tokens: int | None = None, model: str | None = None, reference_type: str | None = None, reference_id: str | None = None, idempotency_key: str | None = None, fixed_job: str | None = None, **extra_data: Any)

Bases: BaseModel

Flexible metadata attached to credit transactions.

Known fields are typed; arbitrary extras pass through to JSONB.

fixed_job : str | None

idempotency_key : str | None

input_tokens : int | None

model : str | None

model_config = {'extra': 'allow'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

output_tokens : int | None

reference_id : str | None

reference_type : str | None

class ducto.interface.models.DailySpendRow(, date: str = '', total_spend: int = 0, transaction_count: int = 0)

Bases: BaseModel

Daily spend aggregation in a time window.

date : str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

total_spend : int

transaction_count : int

class ducto.interface.models.DeductionResult(, transaction_id: str, user_id: str, amount: int, balance_after: int, idempotent: bool = False, error: str | None = None)

Bases: BaseModel

Result of deducting credits after an operation completes.

amount is negative for deductions, positive for refunds.

amount : int

balance_after : int

error : str | None

idempotent : bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

transaction_id : str

user_id : str

class ducto.interface.models.GetUserPlanResult(*, user_id: str, plan_id: str | None = None, plan_name: str | None = None, free_allowance: int = 0, features: dict[str, ~typing.Any]=)

Bases: BaseModel

Result of fetching a user’s current plan.

features : dict[str, Any]

free_allowance : int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

plan_id : str | None

plan_name : str | None

user_id : str

class ducto.interface.models.PlanDefinition(, id: str, name: str, free_allowance: Annotated[int, Ge(ge=0)] = 0, rate_overrides: dict[str, str] | None = None, features: dict[str, Any] | None = None)

Bases: BaseModel

Definition of a subscription plan with free allowance and rate overrides.

features : dict[str, Any] | None

free_allowance : int

id : str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name : str

rate_overrides : dict[str, str] | None

class ducto.interface.models.PricingConfigData(*, version: ~typing.Literal[1] = 1, models: dict[str, str], tools: dict[str, str] = , search: dict[str, str] = , cache: dict[str, str] = , fixed: dict[str, int] = , min_balance: int = 5, plans: dict[str, ~ducto.interface.models.PlanDefinition] | None = None)

Bases: BaseModel

Pricing configuration schema.

Mirrors the YAML config structure used by PricingEngine. Unified format with optional plan definitions.

cache : dict[str, str]

fixed : dict[str, int]

min_balance : int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

models : dict[str, str]

plans : dict[str, PlanDefinition] | None

search : dict[str, str]

tools : dict[str, str]

version : Literal[1]

class ducto.interface.models.PricingConfigHistoryItem(, id: str, version: int, label: str | None = None, active: bool = False, created_at: str = '')

Bases: BaseModel

Lightweight summary for pricing version listing.

active : bool

created_at : str

id : str

label : str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

version : int

class ducto.interface.models.PricingConfigResult(, id: str, config: PricingConfigData, version: int = 1, label: str | None = None)

Bases: BaseModel

Versioned pricing configuration fetched from the store.

config : PricingConfigData

id : str

label : str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

version : int

class ducto.interface.models.RefundResult(, refund_transaction_id: str, original_transaction_id: str, user_id: str, amount: int = 0, new_balance: int = 0, error: str | None = None)

Bases: BaseModel

Result of refunding a credit deduction.

amount : int

error : str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

new_balance : int

original_transaction_id : str

refund_transaction_id : str

user_id : str

class ducto.interface.models.ReserveResult(, reservation_id: str, user_id: str, amount: int, balance: int = 0, reserved_total: int = 0, error: str | None = None)

Bases: BaseModel

Result of reserving credits for an operation.

amount : int

balance : int

error : str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

reservation_id : str

reserved_total : int

user_id : str

class ducto.interface.models.SetUserPlanResult(, user_id: str, plan_id: str)

Bases: BaseModel

Result of assigning a plan to a user.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

plan_id : str

user_id : str

class ducto.interface.models.SetupResult(*, tables_created: list[str] = , rpcs_created: list[str] = , errors: list[str] = )

Bases: BaseModel

Report of what the setup step created or updated.

errors : list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

rpcs_created : list[str]

property success : bool

tables_created : list[str]

class ducto.interface.models.SpendByModelRow(, model: str = '', total_spend: int = 0, transaction_count: int = 0)

Bases: BaseModel

Aggregated spend for a single model in a time window.

model : str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

total_spend : int

transaction_count : int

class ducto.interface.models.SpendByUserRow(, user_id: str = '', total_spend: int = 0, transaction_count: int = 0)

Bases: BaseModel

Aggregated spend for a single user in a time window.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

total_spend : int

transaction_count : int

user_id : str

class ducto.interface.models.SpendCap(, user_id: str = '', type: Literal['daily', 'monthly'] = 'daily', model: str | None = None, limit: Annotated[int, Ge(ge=0)] = 0, action: Literal['deny', 'warn', 'notify'] = 'deny')

Bases: BaseModel

Configuration for a per-user spend cap.

action : Literal['deny', 'warn', 'notify']

cap_type : Literal['daily', 'monthly']

limit : int

model : str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

user_id : str

class ducto.interface.models.SweepResult(, expired_count: int = 0, expired_amount: int = 0, dry_run: bool = False)

Bases: BaseModel

Result of sweeping expired credits.

dry_run : bool

expired_amount : int

expired_count : int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ducto.interface.models.Team(, team_id: str = '', name: str = '', balance: int = 0, member_count: int = 0, created_at: str = '')

Bases: BaseModel

A team with a shared credit balance pool.

balance : int

created_at : str

member_count : int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name : str

team_id : str

class ducto.interface.models.TeamBalanceResult(, team_id: str = '', name: str = '', balance: int = 0, member_count: int = 0)

Bases: BaseModel

Result of fetching team balance.

balance : int

member_count : int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name : str

team_id : str

class ducto.interface.models.TeamDeductionResult(, transaction_id: str = '', team_id: str = '', user_id: str = '', amount: int = 0, team_balance_after: int = 0, error: str | None = None)

Bases: BaseModel

Result of deducting credits from a team pool.

amount : int

error : str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

team_balance_after : int

team_id : str

transaction_id : str

user_id : str

class ducto.interface.models.TeamMember(, user_id: str = '', role: str = '', spend_cap: int | None = None, total_spent: int = 0)

Bases: BaseModel

A member of a team with optional spend cap.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

role : str

spend_cap : int | None

total_spent : int

user_id : str

class ducto.interface.models.TopUserRow(, user_id: str = '', total_spend: int = 0)

Bases: BaseModel

Top-spending user in a time window.

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ducto.interface.models.AddCreditsResult(, transaction_id: str, user_id: str, amount: int, new_balance: int, lifetime_purchased: int = 0)​

amount : int​

lifetime_purchased : int​

model_config = {}​

new_balance : int​

transaction_id : str​

user_id : str​

class ducto.interface.models.AddTeamMemberResult(, team_id: str = '', user_id: str = '', role: str = 'member')​

model_config = {}​

role : str​

team_id : str​

user_id : str​

class ducto.interface.models.AggregateStatsRow(, total_credits_consumed: int = 0, active_users: int = 0, avg_daily_spend: int = 0, top_model: str = '', top_user: str = '')​

active_users : int​

avg_daily_spend : int​

model_config = {}​

top_model : str​

top_user : str​

total_credits_consumed : int​

class ducto.interface.models.AllowanceResult(, plan_id: str, allowance_remaining: int, period_start: str, period_end: str)​

allowance_remaining : int​

model_config = {}​

period_end : str​

period_start : str​

plan_id : str​

class ducto.interface.models.BalanceResult(, user_id: str, balance: int = 0, lifetime_purchased: int = 0)​

balance : int​

lifetime_purchased : int​

model_config = {}​

user_id : str​

class ducto.interface.models.CapCheckResult(, capped: bool = False, current_spend: int = 0, cap_limit: int = 0, action: str | None = None, model: str | None = None)​

action : str | None​

cap_limit : int​

capped : bool​

current_spend : int​

model : str | None​

model_config = {}​

class ducto.interface.models.CheckFeatureResult(, user_id: str, feature: str, value: Any = None, has_feature: bool = False)​

feature : str​

has_feature : bool​

model_config = {}​

user_id : str​

value : Any​

class ducto.interface.models.CreateTeamResult(, team_id: str = '', name: str = '')​

model_config = {}​

name : str​

team_id : str​

class ducto.interface.models.CreditMetadata(, input_tokens: int | None = None, output_tokens: int | None = None, model: str | None = None, reference_type: str | None = None, reference_id: str | None = None, idempotency_key: str | None = None, fixed_job: str | None = None, **extra_data: Any)​

fixed_job : str | None​

idempotency_key : str | None​

input_tokens : int | None​

model : str | None​

model_config = {'extra': 'allow'}​

output_tokens : int | None​

reference_id : str | None​

reference_type : str | None​

class ducto.interface.models.DailySpendRow(, date: str = '', total_spend: int = 0, transaction_count: int = 0)​

date : str​

model_config = {}​

total_spend : int​

transaction_count : int​

class ducto.interface.models.DeductionResult(, transaction_id: str, user_id: str, amount: int, balance_after: int, idempotent: bool = False, error: str | None = None)​

amount : int​

balance_after : int​

error : str | None​

idempotent : bool​

model_config = {}​

transaction_id : str​

user_id : str​

class ducto.interface.models.GetUserPlanResult(*, user_id: str, plan_id: str | None = None, plan_name: str | None = None, free_allowance: int = 0, features: dict[str, ~typing.Any]=)​

features : dict[str, Any]​

free_allowance : int​

model_config = {}​

plan_id : str | None​

plan_name : str | None​

user_id : str​

class ducto.interface.models.PlanDefinition(, id: str, name: str, free_allowance: Annotated[int, Ge(ge=0)] = 0, rate_overrides: dict[str, str] | None = None, features: dict[str, Any] | None = None)​

features : dict[str, Any] | None​

free_allowance : int​

id : str​

class ducto.interface.models.AddCreditsResult(, transaction_id: str, user_id: str, amount: int, new_balance: int, lifetime_purchased: int = 0)

amount : int

lifetime_purchased : int

model_config = {}

new_balance : int

transaction_id : str

user_id : str

class ducto.interface.models.AddTeamMemberResult(, team_id: str = '', user_id: str = '', role: str = 'member')

model_config = {}

role : str

team_id : str

user_id : str

class ducto.interface.models.AggregateStatsRow(, total_credits_consumed: int = 0, active_users: int = 0, avg_daily_spend: int = 0, top_model: str = '', top_user: str = '')

active_users : int

avg_daily_spend : int

model_config = {}

top_model : str

top_user : str

total_credits_consumed : int

class ducto.interface.models.AllowanceResult(, plan_id: str, allowance_remaining: int, period_start: str, period_end: str)

allowance_remaining : int

model_config = {}

period_end : str

period_start : str

plan_id : str

class ducto.interface.models.BalanceResult(, user_id: str, balance: int = 0, lifetime_purchased: int = 0)

balance : int

lifetime_purchased : int

model_config = {}

user_id : str

class ducto.interface.models.CapCheckResult(, capped: bool = False, current_spend: int = 0, cap_limit: int = 0, action: str | None = None, model: str | None = None)

action : str | None

cap_limit : int

capped : bool

current_spend : int

model : str | None

model_config = {}

class ducto.interface.models.CheckFeatureResult(, user_id: str, feature: str, value: Any = None, has_feature: bool = False)

feature : str

has_feature : bool

model_config = {}

user_id : str

value : Any

class ducto.interface.models.CreateTeamResult(, team_id: str = '', name: str = '')

model_config = {}

name : str

team_id : str

class ducto.interface.models.CreditMetadata(, input_tokens: int | None = None, output_tokens: int | None = None, model: str | None = None, reference_type: str | None = None, reference_id: str | None = None, idempotency_key: str | None = None, fixed_job: str | None = None, **extra_data: Any)

fixed_job : str | None

idempotency_key : str | None

input_tokens : int | None

model : str | None

model_config = {'extra': 'allow'}

output_tokens : int | None

reference_id : str | None

reference_type : str | None

class ducto.interface.models.DailySpendRow(, date: str = '', total_spend: int = 0, transaction_count: int = 0)

date : str

model_config = {}

total_spend : int

transaction_count : int

class ducto.interface.models.DeductionResult(, transaction_id: str, user_id: str, amount: int, balance_after: int, idempotent: bool = False, error: str | None = None)

amount : int

balance_after : int

error : str | None

idempotent : bool

model_config = {}

transaction_id : str

user_id : str

class ducto.interface.models.GetUserPlanResult(*, user_id: str, plan_id: str | None = None, plan_name: str | None = None, free_allowance: int = 0, features: dict[str, ~typing.Any]=)

features : dict[str, Any]

free_allowance : int

model_config = {}

plan_id : str | None

plan_name : str | None

user_id : str

class ducto.interface.models.PlanDefinition(, id: str, name: str, free_allowance: Annotated[int, Ge(ge=0)] = 0, rate_overrides: dict[str, str] | None = None, features: dict[str, Any] | None = None)

features : dict[str, Any] | None

free_allowance : int

id : str