Source code for ewoksserver.app.models
from enum import Enum
from pathlib import Path
from typing import Dict, List, Optional
import logging
from pydantic import Field, field_validator
from pydantic import BaseModel
logger = logging.getLogger(__name__)
[docs]
class EwoksSchedulingType(str, Enum):
Local = "local"
Celery = "celery"
[docs]
class EwoksDiscoverySettings(BaseModel):
on_start_up: bool = Field(default=False, title="Discover ewoks tasks on startup")
timeout: Optional[float] = Field(
default=None, title="Timeout for task discovery (in seconds)"
)
[docs]
class EwoksExecutionSettings(BaseModel):
handlers: List[Dict] = Field(default=list(), title="Ewoks execution handlers")
[docs]
class EwoksJobSettings(BaseModel):
type: EwoksSchedulingType = EwoksSchedulingType.Local
configuration: dict = dict()
[docs]
class EwoksSettings(BaseModel):
configured: bool = Field(
default=False, title="Config or resource directory have been defined"
)
resource_directory: Path = Field(
default=Path("."), title="Backend file resource directory"
)
without_events: bool = Field(default=False, title="Enable ewoks events")
ewoks_discovery: EwoksDiscoverySettings = Field(
default=None, title="Ewoks discovery settings", validate_default=True
)
ewoks_execution: EwoksExecutionSettings = Field(
default=None, title="Ewoks execution settings", validate_default=True
)
ewoks_scheduling: EwoksJobSettings = Field(
default=None, title="Ewoks job scheduling settings", validate_default=True
)
[docs]
@field_validator(
"ewoks_discovery", "ewoks_execution", "ewoks_scheduling", mode="before"
)
@classmethod
def set_default_value(cls, input_value):
if input_value is None:
return dict()
return input_value
[docs]
class AppSettings(BaseModel):
no_older_versions: bool = Field(
default=False, title="Do not create end points for older API versions"
)