soulfire.admin.SoulFireAdmin

class soulfire.admin.SoulFireAdmin(*, client: soulfire.client_connect.ClientServiceClient, server: soulfire.server_connect.ServerServiceClient, users: soulfire.user_connect.UserServiceClient, logs: soulfire.logs_connect.LogsServiceClient, metrics: soulfire.metrics_connect.MetricsServiceClient, commands: soulfire.command_connect.CommandServiceClient, downloads: soulfire.download_connect.DownloadServiceClient, plugin_stats: soulfire.plugin_stats_connect.PluginStatsServiceClient, scripts: soulfire.script_connect.ScriptServiceClient, instances: soulfire.instance_connect.InstanceServiceClient)

Async high-level access to SoulFire’s administrative control plane.

client_data(*, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.client_pb2.ClientDataResponse, soulfire.errors.SoulFireOperationError]

generate_webdav_token(*, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[str, soulfire.errors.SoulFireOperationError]

generate_api_token(*, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[str, soulfire.errors.SoulFireOperationError]

update_username(username: str, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError]

update_email(email: str, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError]

invalidate_own_sessions(*, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError]

server_info(*, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.server_pb2.ServerInfoResponse, soulfire.errors.SoulFireOperationError]

update_server_config(config: soulfire.server_pb2.ServerConfig, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError]

set_server_config_entry(request: soulfire.server_pb2.ServerUpdateConfigEntryRequest, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError]

list_users(*, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[list[soulfire.user_pb2.UserListResponse.User], soulfire.errors.SoulFireOperationError]

user(user_id: str, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.user_pb2.UserInfoResponse, soulfire.errors.SoulFireOperationError]

create_user(request: soulfire.user_pb2.UserCreateRequest, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[str, soulfire.errors.SoulFireOperationError]

delete_user(user_id: str, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError]

update_user(request: soulfire.user_pb2.UpdateUserRequest, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError]

invalidate_user_sessions(user_id: str, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError]

generate_user_api_token(user_id: str, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[str, soulfire.errors.SoulFireOperationError]

user_plugin_permission_grants(user_id: str, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[list[soulfire.user_pb2.UserPluginPermissionGrant], soulfire.errors.SoulFireOperationError]

set_user_plugin_permission_grant(user_id: str, permission_id: str, scope: soulfire.plugin_api_pb2.PluginPermissionScope, *, resource_id: str | None = None, granted: bool = True, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.user_pb2.UserPluginPermissionGrant, soulfire.errors.SoulFireOperationError]

delete_user_plugin_permission_grant(user_id: str, permission_id: str, scope: soulfire.plugin_api_pb2.PluginPermissionScope, *, resource_id: str | None = None, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError]

previous_logs(scope: soulfire.logs_pb2.LogScope, count: int = 300, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[list[soulfire.logs_pb2.LogString], soulfire.errors.SoulFireOperationError]

logs(scope: soulfire.logs_pb2.LogScope, *, headers: Headers = None, timeout_ms: int | None = None) → soulfire.streams.Stream[soulfire.logs_pb2.LogResponse, soulfire.errors.SoulFireOperationError]

server_metrics(since: google.protobuf.timestamp_pb2.Timestamp | None = None, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.metrics_pb2.GetServerMetricsResponse, soulfire.errors.SoulFireOperationError]

instance_metrics(instance_id: str, since: google.protobuf.timestamp_pb2.Timestamp | None = None, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.metrics_pb2.GetInstanceMetricsResponse, soulfire.errors.SoulFireOperationError]

execute_command(request: soulfire.command_pb2.CommandRequest, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.command_pb2.CommandResponse, soulfire.errors.SoulFireOperationError]

complete_command(request: soulfire.command_pb2.CommandCompletionRequest, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.command_pb2.CommandCompletionResponse, soulfire.errors.SoulFireOperationError]

download(request: soulfire.download_pb2.DownloadRequest, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.download_pb2.DownloadResponse, soulfire.errors.SoulFireOperationError]

plugin_stats(instance_id: str, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[list[soulfire.plugin_stats_pb2.PluginRuntimeStat], soulfire.errors.SoulFireOperationError]

audit_log(instance_id: str, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[list[soulfire.instance_pb2.InstanceAuditLogResponse.AuditLogEntry], soulfire.errors.SoulFireOperationError]

list_scripts(instance_id: str, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[list[soulfire.script_pb2.ScriptInfo], soulfire.errors.SoulFireOperationError]

script(instance_id: str, script_id: str, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.script_pb2.GetScriptResponse, soulfire.errors.SoulFireOperationError]

create_script(request: soulfire.script_pb2.CreateScriptRequest, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.script_pb2.CreateScriptResponse, soulfire.errors.SoulFireOperationError]

update_script(request: soulfire.script_pb2.UpdateScriptRequest, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.script_pb2.UpdateScriptResponse, soulfire.errors.SoulFireOperationError]

delete_script(instance_id: str, script_id: str, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError]

activate_script(instance_id: str, script_id: str, *, headers: Headers = None, timeout_ms: int | None = None) → soulfire.streams.Stream[soulfire.script_pb2.ScriptEvent, soulfire.errors.SoulFireOperationError]

deactivate_script(instance_id: str, script_id: str, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[None, soulfire.errors.SoulFireOperationError]

script_status(instance_id: str, script_id: str, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.script_pb2.GetScriptStatusResponse, soulfire.errors.SoulFireOperationError]

script_logs(request: soulfire.script_pb2.SubscribeScriptLogsRequest, *, headers: Headers = None, timeout_ms: int | None = None) → soulfire.streams.Stream[soulfire.script_pb2.ScriptLogEntry, soulfire.errors.SoulFireOperationError]

node_types(request: soulfire.script_pb2.GetNodeTypesRequest | None = None, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.script_pb2.GetNodeTypesResponse, soulfire.errors.SoulFireOperationError]

script_registry_data(request: soulfire.script_pb2.GetRegistryDataRequest | None = None, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.script_pb2.GetRegistryDataResponse, soulfire.errors.SoulFireOperationError]

validate_script(request: soulfire.script_pb2.ValidateScriptRequest, *, headers: Headers = None, timeout_ms: int | None = None) → effect_py.Effect[soulfire.script_pb2.ValidateScriptResponse, soulfire.errors.SoulFireOperationError]

dry_run_script(request: soulfire.script_pb2.DryRunScriptRequest, *, headers: Headers = None, timeout_ms: int | None = None) → soulfire.streams.Stream[soulfire.script_pb2.ScriptEvent, soulfire.errors.SoulFireOperationError]