dataflow.server_management.base¶

Generic base server manager with full type safety.

This module provides the abstract base class for all server managers in Haive. It uses Python generics to ensure type safety while providing common functionality.

Classes¶

BaseServerManager

Generic base class for all server managers.

ServerLifecycle

Protocol for server lifecycle operations.

Module Contents¶

class dataflow.server_management.base.BaseServerManager(/, **data)¶

Bases: pydantic.BaseModel, abc.ABC, Generic[ConfigT, InfoT]

Generic base class for all server managers.

This class provides a type-safe foundation for managing servers of any type. Subclasses should specify the concrete ConfigT and InfoT types.

Type Parameters:

ConfigT: Server configuration type (must extend BaseServerConfig) InfoT: Server runtime info type (must extend BaseServerInfo)

Parameters:

data (Any)

servers¶

Currently running servers mapped by name

available_configs¶

Available server configurations

config_class¶

Configuration class for type validation

info_class¶

Info class for runtime information

auto_restart¶

Whether to automatically restart failed servers

max_restart_attempts¶

Maximum restart attempts before giving up

health_check_interval¶

Seconds between health checks

restart_tracking¶

Tracks restart attempts per server

Example

Creating a concrete server manager:

class MyServerManager(BaseServerManager[MyConfig, MyInfo]):
    config_class = Field(default=MyConfig, exclude=True)
    info_class = Field(default=MyInfo, exclude=True)

    async def start_server(self, name: str, config: Optional[MyConfig] = None) -> MyInfo:
        # Implementation
        pass

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

add_config(name, config)¶

Add or update a server configuration.

Parameters:

  • name (str) – Server name

  • config (Union[ConfigT, Dict[str, Any]]) – Configuration object or dict

Returns:

Validated configuration object

Raises:

ValidationError – If configuration is invalid

Return type:

ConfigT

async cleanup()¶

Clean up resources and stop all servers.

Return type:

None

get_config(name)¶

Get server configuration by name.

Parameters:

name (str) – Server name

Returns:

Configuration object or None if not found

Return type:

Optional[ConfigT]

get_server_info(name)¶

Get runtime information for a server.

Parameters:

name (str) – Server name

Returns:

Server info or None if not running

Return type:

Optional[InfoT]

get_stats()¶

Get server manager statistics.

Returns:

  • total_configured: Number of configured servers

  • total_running: Number of running servers

  • servers_by_status: Count by status

  • restart_counts: Restart attempts per server

Return type:

Dictionary with stats including

abstractmethod health_check(name)¶
Async:

Parameters:

name (str)

Return type:

bool

Check if server is healthy.

Parameters:

name (str) – Server name

Returns:

True if healthy, False otherwise

Return type:

bool

is_running(name)¶

Check if server is running.

Parameters:

name (str) – Server name

Returns:

True if server is running

Return type:

bool

list_servers(status=None)¶

List servers by status.

Parameters:

status (Optional[dataflow.server_management.models.ServerStatus]) – Optional status filter

Returns:

List of server names

Return type:

List[str]

model_post_init(__context)¶

Additional initialization after validation.

Parameters:

__context (Any)

Return type:

None

remove_config(name)¶

Remove a server configuration.

Parameters:

name (str) – Server name

Returns:

True if removed, False if not found

Raises:

RuntimeError – If server is currently running

Return type:

bool

abstractmethod restart_server(name)¶
Async:

Parameters:

name (str)

Return type:

InfoT

Restart a server.

Parameters:

name (str) – Server name

Returns:

New server runtime information

Return type:

InfoT

async start_health_monitoring(name)¶

Start health monitoring for a server.

Parameters:

name (str) – Server name

Return type:

None

abstractmethod start_server(name, config=None)¶
Async:

Parameters:

  • name (str)

  • config (Optional[ConfigT])

Return type:

InfoT

Start a server with given configuration.

Parameters:

  • name (str) – Server name

  • config (Optional[ConfigT]) – Optional configuration override

Returns:

Server runtime information

Raises:
Return type:

InfoT

async stop_health_monitoring(name)¶

Stop health monitoring for a server.

Parameters:

name (str) – Server name

Return type:

None

abstractmethod stop_server(name, force=False)¶
Async:

Parameters:
Return type:

bool

Stop a running server.

Parameters:

  • name (str) – Server name

  • force (bool) – Force stop if true

Returns:

True if stopped successfully

Return type:

bool

classmethod validate_config_types(v, info)¶

Ensure all configs are the correct type.

Parameters:

  • v (Dict[str, Any])

  • info (pydantic.ValidationInfo)

Return type:

Dict[str, Any]

classmethod validate_info_types(v, info)¶

Ensure all server info objects are the correct type.

Parameters:

  • v (Dict[str, Any])

  • info (pydantic.ValidationInfo)

Return type:

Dict[str, Any]

validate_restart_policy()¶

Validate restart configuration consistency.

Return type:

BaseServerManager

validate_server_consistency()¶

Ensure running servers have configurations.

Return type:

BaseServerManager

model_config¶

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

class dataflow.server_management.base.ServerLifecycle¶

Bases: Protocol

Protocol for server lifecycle operations.

async health_check(name)¶

Check if server is healthy.

Parameters:

name (str)

Return type:

bool

async restart_server(name)¶

Restart a server.

Parameters:

name (str)

Return type:

Any

async start_server(name, config=None)¶

Start a server with given configuration.

Parameters:

  • name (str)

  • config (Optional[Any])

Return type:

Any

async stop_server(name, force=False)¶

Stop a running server.

Parameters:
Return type:

bool