dataflow.server_management.models

Base models for server management.

This module defines the fundamental Pydantic models used for server configuration and runtime information across all server types.

Attributes

logger

Classes

BaseServerConfig	Base configuration all servers must have.
BaseServerInfo	Base runtime info all servers have.
ServerStatus	Universal server status enum.

Module Contents

class dataflow.server_management.models.BaseServerConfig(/, **data: Any)

Bases: pydantic.BaseModel

Base configuration all servers must have.

This model defines the minimum configuration required for any server type. Subclasses should extend this with type-specific fields.

name

Unique server identifier

command

Command to execute (executable + args)

description

Human-readable description

working_directory

Optional working directory for process

environment

Additional environment variables

timeout_seconds

Startup timeout in seconds

auto_restart

Whether to restart on failure

health_check_enabled

Whether to perform health checks

Example

Basic server configuration:

config = BaseServerConfig(
    name="my-server",
    command=["python", "-m", "server"],
    description="My custom server"
)

classmethod validate_command_not_empty(v: List[str]) → List[str]

Validate command has executable and no empty strings.

classmethod validate_environment_vars(v: Dict[str, str]) → Dict[str, str]

Validate environment variable names and values.

classmethod validate_name_format(v: str) → str

Validate name follows naming convention.

classmethod validate_working_directory(v: str | None) → str | None

Validate working directory if provided.

auto_restart: bool = None

command: List[str] = None

description: str = None

environment: Dict[str, str] = None

health_check_enabled: bool = None

model_config

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

name: str = None

timeout_seconds: int = None

working_directory: str | None = None

class dataflow.server_management.models.BaseServerInfo(/, **data: Any)

Bases: pydantic.BaseModel

Base runtime info all servers have.

This model represents the runtime state of a server process. It excludes the actual process handle from serialization.

name

Server name

pid

Process ID

status

Current server status

started_at

When the server was started

config_snapshot

Configuration used to start server

error_message

Last error message if any

restart_count

Number of times server has been restarted

last_health_check

Last successful health check time

uptime_seconds

Computed uptime in seconds

record_health_check(success: bool) → None

Record health check result.

update_status(new_status: ServerStatus, error: str | None = None) → None

Update server status with optional error message.

config_snapshot: Dict[str, Any] = None

error_message: str | None = None

property is_running: bool

Check if server process is currently running.

last_health_check: datetime.datetime | None = None

model_config

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

name: str = None

pid: int = None

process_handle: Any | None = None

restart_count: int = None

started_at: datetime.datetime = None

status: ServerStatus = None

property uptime_seconds: float

Calculate server uptime in seconds.

class dataflow.server_management.models.ServerStatus

Bases: str, enum.Enum

Universal server status enum.

ERROR = 'error'

HEALTH_CHECK_FAILED = 'health_check_failed'

RESTARTING = 'restarting'

RUNNING = 'running'

STARTING = 'starting'

STOPPED = 'stopped'

STOPPING = 'stopping'

dataflow.server_management.models.logger