chore: remove unused generated models, expose git exceptions from wrenn.exceptions

This commit is contained in:
2026-05-23 17:39:51 +06:00
parent 7b04ba3949
commit 4dcbc73003
3 changed files with 27 additions and 742 deletions

View File

@ -164,4 +164,17 @@ def __getattr__(name: str) -> type:
stacklevel=2, stacklevel=2,
) )
return WrennHostHasCapsulesError return WrennHostHasCapsulesError
if name in ("GitError", "GitCommandError", "GitAuthError"):
from wrenn._git.exceptions import (
GitAuthError as _GitAuthError,
GitCommandError as _GitCommandError,
GitError as _GitError,
)
_m: dict[str, type] = {
"GitError": _GitError,
"GitCommandError": _GitCommandError,
"GitAuthError": _GitAuthError,
}
return _m[name]
raise AttributeError(f"module {__name__!r} has no attribute {name!r}") raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

View File

@ -1,63 +1,17 @@
from wrenn.models._generated import ( from wrenn.models._generated import (
APIKeyResponse,
Capsule, Capsule,
CreateAPIKeyRequest,
CreateCapsuleRequest,
CreateHostRequest,
CreateHostResponse,
CreateSnapshotRequest,
Encoding,
Error,
Error1,
ExecRequest,
ExecResponse,
FileEntry, FileEntry,
Host,
ListDirRequest,
ListDirResponse, ListDirResponse,
LoginRequest,
MakeDirRequest,
MakeDirResponse, MakeDirResponse,
ReadFileRequest,
RegisterHostRequest,
RegisterHostResponse,
RemoveRequest,
SignupRequest,
Status, Status,
Status1,
Template, Template,
Type,
Type2,
) )
__all__ = [ __all__ = [
"APIKeyResponse",
"CreateAPIKeyRequest",
"CreateHostRequest",
"CreateHostResponse",
"CreateCapsuleRequest",
"CreateSnapshotRequest",
"Encoding",
"Error",
"Error1",
"ExecRequest",
"ExecResponse",
"FileEntry",
"Host",
"ListDirRequest",
"ListDirResponse",
"LoginRequest",
"MakeDirRequest",
"MakeDirResponse",
"ReadFileRequest",
"RegisterHostRequest",
"RegisterHostResponse",
"RemoveRequest",
"Capsule", "Capsule",
"SignupRequest", "FileEntry",
"ListDirResponse",
"MakeDirResponse",
"Status", "Status",
"Status1",
"Template", "Template",
"Type",
"Type2",
] ]

View File

@ -1,147 +1,13 @@
# generated by datamodel-codegen: # generated by datamodel-codegen:
# filename: openapi.yaml # filename: openapi.yaml
# timestamp: 2026-05-22T19:20:45+00:00 # timestamp: 2026-05-23T11:20:02+00:00
from __future__ import annotations from __future__ import annotations
from pydantic import AwareDatetime, BaseModel, EmailStr, Field from pydantic import AwareDatetime, BaseModel, Field
from typing import Annotated, Any from typing import Annotated
from datetime import date as date_aliased
from enum import StrEnum from enum import StrEnum
class SignupRequest(BaseModel):
email: EmailStr
password: Annotated[str, Field(min_length=8)]
name: Annotated[str, Field(max_length=100)]
class LoginRequest(BaseModel):
email: EmailStr
password: str
class SignupResponse(BaseModel):
message: Annotated[
str | None,
Field(description="Confirmation message instructing user to check email"),
] = None
class SessionResponse(BaseModel):
"""
Returned by login, activate, and switch-team. The actual auth credential
is the wrenn_sid cookie set on the response. The body carries identity
data the SPA needs to bootstrap.
"""
user_id: str | None = None
team_id: str | None = None
email: str | None = None
name: str | None = None
role: str | None = None
is_admin: bool | None = None
class CreateAPIKeyRequest(BaseModel):
name: str | None = "Unnamed API Key"
class APIKeyResponse(BaseModel):
id: str | None = None
team_id: str | None = None
name: str | None = None
key_prefix: Annotated[
str | None, Field(description='Display prefix (e.g. "wrn_ab12cd34...")')
] = None
created_at: AwareDatetime | None = None
last_used: AwareDatetime | None = None
key: Annotated[
str | None,
Field(
description="Full plaintext key. Only returned on creation, never again."
),
] = None
class CreateCapsuleRequest(BaseModel):
template: str | None = "minimal-ubuntu"
vcpus: int | None = 1
memory_mb: int | None = 512
disk_size_mb: Annotated[
int | None,
Field(
description="Maximum size of the per-capsule copy-on-write disk in MB. Capped at 5 GB by default; the actual size is max(disk_size_mb, origin rootfs size).\n"
),
] = 5120
timeout_sec: Annotated[
int | None,
Field(
description="Auto-pause TTL in seconds. The capsule is automatically paused after this duration of inactivity (no exec or ping). 0 means no auto-pause. Positive values below 60 are silently clamped to 60 (the agent's startup envelope).\n",
ge=0,
),
] = 0
class Point(BaseModel):
date: date_aliased | None = None
cpu_minutes: float | None = None
ram_mb_minutes: float | None = None
class UsageResponse(BaseModel):
from_: Annotated[date_aliased | None, Field(alias="from")] = None
to: date_aliased | None = None
points: list[Point] | None = None
class Range(StrEnum):
field_5m = "5m"
field_1h = "1h"
field_6h = "6h"
field_24h = "24h"
field_30d = "30d"
class Current(BaseModel):
running_count: int | None = None
vcpus_reserved: int | None = None
memory_mb_reserved: int | None = None
sampled_at: AwareDatetime | None = None
class Peaks(BaseModel):
"""
Maximum values over the last 30 days.
"""
running_count: int | None = None
vcpus: int | None = None
memory_mb: int | None = None
class Series(BaseModel):
"""
Parallel arrays for chart rendering.
"""
labels: list[AwareDatetime] | None = None
running: list[int] | None = None
vcpus: list[int] | None = None
memory_mb: list[int] | None = None
class CapsuleStats(BaseModel):
range: Range | None = None
current: Current | None = None
peaks: Annotated[
Peaks | None, Field(description="Maximum values over the last 30 days.")
] = None
series: Annotated[
Series | None, Field(description="Parallel arrays for chart rendering.")
] = None
class Status(StrEnum): class Status(StrEnum):
pending = "pending" pending = "pending"
starting = "starting" starting = "starting"
@ -164,8 +30,6 @@ class Capsule(BaseModel):
vcpus: int | None = None vcpus: int | None = None
memory_mb: int | None = None memory_mb: int | None = None
timeout_sec: int | None = None timeout_sec: int | None = None
guest_ip: str | None = None
host_ip: str | None = None
created_at: AwareDatetime | None = None created_at: AwareDatetime | None = None
started_at: AwareDatetime | None = None started_at: AwareDatetime | None = None
last_active_at: AwareDatetime | None = None last_active_at: AwareDatetime | None = None
@ -176,16 +40,14 @@ class Capsule(BaseModel):
description="Free-form key/value labels attached at create-time. Also carries\nagent-side version info (kernel_version, vmm_version,\nagent_version, envd_version) when running.\n" description="Free-form key/value labels attached at create-time. Also carries\nagent-side version info (kernel_version, vmm_version,\nagent_version, envd_version) when running.\n"
), ),
] = None ] = None
disk_size_mb: int | None = None disk_size_mb: Annotated[
int | None, Field(description="Maximum disk capacity in MiB.")
] = None
class CreateSnapshotRequest(BaseModel): disk_used_mb: Annotated[
sandbox_id: Annotated[ int | None,
str, Field(description="ID of the running capsule to snapshot.") Field(
] description="Current disk usage in MiB. Only populated on individual capsule GET; omitted in list responses."
name: Annotated[ ),
str | None,
Field(description="Name for the snapshot template. Auto-generated if omitted."),
] = None ] = None
@ -216,121 +78,6 @@ class Template(BaseModel):
metadata: dict[str, str] | None = None metadata: dict[str, str] | None = None
class AdminTemplate(BaseModel):
"""
Template as returned by the admin templates list. Unlike `Template`
(the team-facing snapshot shape), this includes the owning `team_id`
and omits `platform`/`metadata`.
"""
name: str | None = None
type: Type | None = None
vcpus: int | None = None
memory_mb: int | None = None
size_bytes: int | None = None
team_id: Annotated[
str | None,
Field(
description="Owning team ID (formatted, e.g. `team-…`). Platform team for global templates."
),
] = None
created_at: AwareDatetime | None = None
protected: Annotated[
bool | None,
Field(
description="True for built-in system base templates (minimal-ubuntu,\nminimal-alpine, minimal-arch, minimal-fedora). Protected templates\ncannot be deleted.\n"
),
] = None
class ExecRequest(BaseModel):
cmd: str
args: list[str] | None = None
timeout_sec: Annotated[
int | None,
Field(description="Timeout in seconds (foreground exec only, default 30)"),
] = 30
background: Annotated[
bool | None,
Field(
description="If true, starts the process in the background and returns immediately with a PID and tag (HTTP 202)"
),
] = False
tag: Annotated[
str | None,
Field(
description="Optional user-chosen tag for the background process. Auto-generated if omitted. Only used when background is true."
),
] = None
envs: Annotated[
dict[str, str] | None,
Field(
description="Environment variables for the process (background exec only)"
),
] = None
cwd: Annotated[
str | None,
Field(description="Working directory for the process (background exec only)"),
] = None
class BackgroundExecResponse(BaseModel):
sandbox_id: str | None = None
cmd: str | None = None
pid: int | None = None
tag: str | None = None
class ProcessEntry(BaseModel):
pid: int | None = None
tag: str | None = None
cmd: str | None = None
args: list[str] | None = None
class ProcessListResponse(BaseModel):
processes: list[ProcessEntry] | None = None
class Encoding(StrEnum):
"""
Output encoding. "base64" when stdout/stderr contain binary data.
"""
utf_8 = "utf-8"
base64 = "base64"
class ExecResponse(BaseModel):
sandbox_id: str | None = None
cmd: str | None = None
stdout: str | None = None
stderr: str | None = None
exit_code: int | None = None
duration_ms: int | None = None
encoding: Annotated[
Encoding | None,
Field(
description='Output encoding. "base64" when stdout/stderr contain binary data.'
),
] = None
class ReadFileRequest(BaseModel):
path: Annotated[str, Field(description="Absolute file path inside the capsule")]
class ListDirRequest(BaseModel):
path: Annotated[str, Field(description="Directory path inside the capsule")]
depth: Annotated[
int | None,
Field(
description="Recursion depth (0 = non-recursive, 1 = immediate children)"
),
] = 1
class Type2(StrEnum): class Type2(StrEnum):
file = "file" file = "file"
directory = "directory" directory = "directory"
@ -354,438 +101,9 @@ class FileEntry(BaseModel):
symlink_target: str | None = None symlink_target: str | None = None
class MakeDirRequest(BaseModel):
path: Annotated[
str, Field(description="Directory path to create inside the capsule")
]
class MakeDirResponse(BaseModel): class MakeDirResponse(BaseModel):
entry: FileEntry | None = None entry: FileEntry | None = None
class RemoveRequest(BaseModel):
path: Annotated[str, Field(description="Path to remove inside the capsule")]
class Type3(StrEnum):
"""
Host type. Regular hosts are shared; BYOC hosts belong to a team.
"""
regular = "regular"
byoc = "byoc"
class CreateHostRequest(BaseModel):
type: Annotated[
Type3,
Field(
description="Host type. Regular hosts are shared; BYOC hosts belong to a team."
),
]
team_id: Annotated[str | None, Field(description="Required for BYOC hosts.")] = None
provider: Annotated[
str | None,
Field(description="Cloud provider (e.g. aws, gcp, hetzner, bare-metal)."),
] = None
availability_zone: Annotated[
str | None, Field(description="Availability zone (e.g. us-east, eu-west).")
] = None
class RegisterHostRequest(BaseModel):
token: Annotated[
str, Field(description="One-time registration token from POST /v1/hosts.")
]
arch: Annotated[
str | None, Field(description="CPU architecture (e.g. x86_64, aarch64).")
] = None
cpu_cores: int | None = None
memory_mb: int | None = None
disk_gb: int | None = None
address: Annotated[str, Field(description="Host agent address (ip:port).")]
class Type4(StrEnum):
regular = "regular"
byoc = "byoc"
class Status1(StrEnum):
pending = "pending"
online = "online"
offline = "offline"
draining = "draining"
unreachable = "unreachable"
class Host(BaseModel):
id: str | None = None
type: Type4 | None = None
team_id: str | None = None
provider: str | None = None
availability_zone: str | None = None
arch: str | None = None
cpu_cores: int | None = None
memory_mb: int | None = None
disk_gb: int | None = None
address: str | None = None
status: Status1 | None = None
last_heartbeat_at: AwareDatetime | None = None
created_by: str | None = None
created_at: AwareDatetime | None = None
updated_at: AwareDatetime | None = None
class RefreshHostTokenRequest(BaseModel):
refresh_token: Annotated[
str,
Field(
description="Refresh token obtained from registration or a previous refresh."
),
]
class RefreshHostTokenResponse(BaseModel):
host: Host | None = None
token: Annotated[
str | None, Field(description="New host JWT. Valid for 7 days.")
] = None
refresh_token: Annotated[
str | None,
Field(
description="New refresh token. Valid for 60 days; old token is revoked."
),
] = None
class HostDeletePreview(BaseModel):
host: Host | None = None
sandbox_ids: Annotated[
list[str] | None,
Field(description="IDs of capsules that would be destroyed on force-delete."),
] = None
class Error(BaseModel):
code: Annotated[str | None, Field(examples=["host_has_sandboxes"])] = None
message: str | None = None
sandbox_ids: Annotated[
list[str] | None, Field(description="IDs of active capsules blocking deletion.")
] = None
class HostHasCapsulesError(BaseModel):
error: Error | None = None
class AddTagRequest(BaseModel):
tag: str
class UserSearchResult(BaseModel):
user_id: str | None = None
email: str | None = None
class Team(BaseModel):
id: str | None = None
name: str | None = None
slug: Annotated[
str | None, Field(description="Immutable 12-char hex slug (e.g. a1b2c3-d1e2f3)")
] = None
created_at: AwareDatetime | None = None
class Role(StrEnum):
owner = "owner"
admin = "admin"
member = "member"
class TeamWithRole(Team):
role: Role | None = None
class TeamMember(BaseModel):
user_id: str | None = None
email: str | None = None
role: Role | None = None
joined_at: AwareDatetime | None = None
class TeamDetail(BaseModel):
team: Team | None = None
members: list[TeamMember] | None = None
class Range1(StrEnum):
field_5m = "5m"
field_10m = "10m"
field_1h = "1h"
field_2h = "2h"
field_6h = "6h"
field_12h = "12h"
field_24h = "24h"
class MetricPoint(BaseModel):
timestamp_unix: int | None = None
cpu_pct: Annotated[
float | None,
Field(
description="CPU utilization percentage (0-100), normalized to vCPU count"
),
] = None
mem_bytes: Annotated[
int | None,
Field(
description="Resident memory in bytes (VmRSS of Cloud Hypervisor process)"
),
] = None
disk_bytes: Annotated[
int | None, Field(description="Allocated disk bytes for the CoW sparse file")
] = None
class Provider(StrEnum):
discord = "discord"
slack = "slack"
teams = "teams"
googlechat = "googlechat"
telegram = "telegram"
matrix = "matrix"
webhook = "webhook"
class Event(StrEnum):
capsule_create = "capsule.create"
capsule_pause = "capsule.pause"
capsule_resume = "capsule.resume"
capsule_destroy = "capsule.destroy"
template_snapshot_create = "template.snapshot.create"
template_snapshot_delete = "template.snapshot.delete"
host_up = "host.up"
host_down = "host.down"
class CreateChannelRequest(BaseModel):
name: Annotated[str, Field(description="Unique channel name within the team.")]
provider: Provider
config: Annotated[
dict[str, str],
Field(
description='Provider-specific configuration fields. Discord/Slack/Teams/Google Chat: {"webhook_url": "..."}. Telegram: {"bot_token": "...", "chat_id": "..."}. Matrix: {"homeserver_url": "...", "access_token": "...", "room_id": "..."}. Webhook: {"url": "...", "secret": "..."} (secret is auto-generated if omitted).\n'
),
]
events: list[Event]
class TestChannelRequest(BaseModel):
provider: Provider
config: Annotated[
dict[str, str],
Field(
description="Provider-specific configuration fields (same as CreateChannelRequest.config)."
),
]
class RotateConfigRequest(BaseModel):
config: Annotated[
dict[str, str],
Field(
description="New provider configuration fields. Must include all required fields for the channel's provider. Replaces the existing config entirely.\n"
),
]
class UpdateChannelRequest(BaseModel):
name: str
events: list[Event]
class ChannelResponse(BaseModel):
id: str | None = None
team_id: str | None = None
name: str | None = None
provider: Provider | None = None
events: list[str] | None = None
created_at: AwareDatetime | None = None
updated_at: AwareDatetime | None = None
secret: Annotated[
str | None,
Field(description="Webhook secret. Only returned on creation, never again."),
] = None
class MeResponse(BaseModel):
name: str | None = None
email: EmailStr | None = None
has_password: Annotated[
bool | None,
Field(
description="Whether the user has a password set (false for OAuth-only accounts)"
),
] = None
providers: Annotated[
list[str] | None,
Field(description='List of linked OAuth provider names (e.g. ["github"])'),
] = None
class ChangePasswordRequest(BaseModel):
current_password: Annotated[
str | None, Field(description="Required when changing an existing password")
] = None
new_password: Annotated[str, Field(min_length=8)]
confirm_password: Annotated[
str | None,
Field(
description="Required when adding a password to an OAuth-only account (must match new_password)"
),
] = None
class Error2(BaseModel):
code: str | None = None
message: str | None = None
class Error1(BaseModel):
error: Error2 | None = None
class ActorType(StrEnum):
user = "user"
api_key = "api_key"
host = "host"
system = "system"
class Status2(StrEnum):
success = "success"
failure = "failure"
class AuditLogEntry(BaseModel):
id: str | None = None
actor_type: ActorType | None = None
actor_id: str | None = None
actor_name: str | None = None
resource_type: str | None = None
resource_id: str | None = None
action: str | None = None
scope: str | None = None
status: Status2 | None = None
metadata: dict[str, Any] | None = None
created_at: AwareDatetime | None = None
class Event2(StrEnum):
connected = "connected"
capsule_create = "capsule.create"
capsule_pause = "capsule.pause"
capsule_resume = "capsule.resume"
capsule_destroy = "capsule.destroy"
capsule_state_changed = "capsule.state.changed"
template_snapshot_create = "template.snapshot.create"
template_snapshot_delete = "template.snapshot.delete"
host_up = "host.up"
host_down = "host.down"
class Outcome(StrEnum):
"""
Present for action events (capsule.* except state.changed,
template.snapshot.*). Absent for host.up/down, capsule.state.changed,
and the connected sentinel.
"""
success = "success"
error = "error"
class Resource(BaseModel):
id: str | None = None
type: str | None = None
class Type5(StrEnum):
user = "user"
api_key = "api_key"
system = "system"
class Actor(BaseModel):
type: Type5 | None = None
id: str | None = None
name: str | None = None
class SSEEvent(BaseModel):
"""
Wire format of one SSE message body. The event name (`event:` line) is
the `kind` and the JSON below is the `data:` line.
"""
event: Event2 | None = None
outcome: Annotated[
Outcome | None,
Field(
description="Present for action events (capsule.* except state.changed,\ntemplate.snapshot.*). Absent for host.up/down, capsule.state.changed,\nand the connected sentinel.\n"
),
] = None
resource: Resource | None = None
actor: Actor | None = None
metadata: Annotated[
dict[str, str] | None,
Field(
description="Event-specific context. Examples: `reason` (ttl_expired,\nhost_failure, cleanup_after_create_error, orphaned),\n`host_ip`, `from`/`to` (for capsule.state.changed).\n"
),
] = None
error: Annotated[
str | None, Field(description="Failure reason; only set when outcome=error.")
] = None
sandbox: Annotated[
Capsule | None,
Field(description="Populated for capsule.* events; null if DB lookup failed."),
] = None
timestamp: AwareDatetime | None = None
class ListDirResponse(BaseModel): class ListDirResponse(BaseModel):
entries: list[FileEntry] | None = None entries: list[FileEntry] | None = None
class CreateHostResponse(BaseModel):
host: Host | None = None
registration_token: Annotated[
str | None,
Field(
description="One-time registration token for the host agent. Expires in 1 hour."
),
] = None
class RegisterHostResponse(BaseModel):
host: Host | None = None
token: Annotated[
str | None,
Field(description="Host JWT for X-Host-Token header. Valid for 7 days."),
] = None
refresh_token: Annotated[
str | None,
Field(
description="Refresh token for obtaining new JWTs. Valid for 60 days; rotated on each use."
),
] = None
class CapsuleMetrics(BaseModel):
sandbox_id: str | None = None
range: Range1 | None = None
points: list[MetricPoint] | None = None