flip_api.domain.interfaces.fl

Attributes

`REQUIRED_JOB_TYPES_FILE`
`_JOB_TYPES_CONFIG`
`JobTypes`

Classes

`IStartTrainingBody`
`ISchedulerResponse`
`IJobResponse`
`IJobMetaData`	Job metadata as returned by an FL-API adapter's `GET /list_jobs`.
`IRequiredTrainingInformation`
`IInitiateTrainingInputPayload`
`INetDetails`
`IServerStatus`	Defines the status of the server.
`IClientStatus`	Defines the status of a client.
`INetStatus`
`IOverridableConfig`
`FLAggregators`	Enumeration for different FL aggregators
`AggregationWeights`
`JobRequiredFiles`

Functions

_load_job_types_config(→ dict[str, list[str]])

Loads the job types configuration from the JSON file.

Module Contents

flip_api.domain.interfaces.fl.REQUIRED_JOB_TYPES_FILE

flip_api.domain.interfaces.fl._load_job_types_config() → dict[str, list[str]]

Loads the job types configuration from the JSON file.

Returns:

A dictionary mapping job type names to their required files.

Return type:

dict[str, list[str]]

Raises:

FileNotFoundError – If the JOB_TYPES_REQUIRED_FILES_FILE file is not found.
json.JSONDecodeError – If the JSON file is malformed.

flip_api.domain.interfaces.fl._JOB_TYPES_CONFIG

class flip_api.domain.interfaces.fl.IStartTrainingBody

Bases: pydantic.BaseModel

project_id: str

cohort_query: str

trusts: list[str]

bundle_urls: list[str]

class flip_api.domain.interfaces.fl.ISchedulerResponse

Bases: pydantic.BaseModel

id: uuid.UUID

netId: uuid.UUID

class flip_api.domain.interfaces.fl.IJobResponse

Bases: pydantic.BaseModel

id: uuid.UUID

model_id: uuid.UUID

clients: list[str]

class flip_api.domain.interfaces.fl.IJobMetaData

Bases: pydantic.BaseModel

Job metadata as returned by an FL-API adapter’s GET /list_jobs.

The shared job-metadata contract (GitHub issue #490). flip-api correlates model_id <-> job_id in its own fl_job table, so the contract carries only job_id + status.

model_config

job_id: str

status: flip_api.domain.schemas.status.FLJobStatus

class flip_api.domain.interfaces.fl.IRequiredTrainingInformation

Bases: pydantic.BaseModel

project_id: str

cohort_query: str

class flip_api.domain.interfaces.fl.IInitiateTrainingInputPayload

Bases: pydantic.BaseModel

trusts: list[flip_api.domain.schemas.types.TrimStr]

classmethod must_be_unique(v)

class flip_api.domain.interfaces.fl.INetDetails

Bases: pydantic.BaseModel

name: str

endpoint: str

class flip_api.domain.interfaces.fl.IServerStatus

Bases: pydantic.BaseModel

Defines the status of the server.

model_config

status: str

class flip_api.domain.interfaces.fl.IClientStatus

Bases: pydantic.BaseModel

Defines the status of a client.

model_config

name: str

status: str

property online: bool

class flip_api.domain.interfaces.fl.INetStatus

Bases: pydantic.BaseModel

name: str

fl_backend: Literal['nvflare', 'flower']

online: bool | None = None

registered_clients: int | None = None

net_in_use: bool | None = None

clients: list[IClientStatus]

class flip_api.domain.interfaces.fl.IOverridableConfig

Bases: pydantic.BaseModel

LOCAL_ROUNDS: int | None = None

GLOBAL_ROUNDS: int | None = None

IGNORE_RESULT_ERROR: bool | None = None

AGGREGATOR: str | None = None

AGGREGATION_WEIGHTS: dict[str, float] | None = None

class flip_api.domain.interfaces.fl.FLAggregators(*args, **kwds)

Bases: enum.Enum

Enumeration for different FL aggregators

InTimeAccumulateWeightedAggregator = 'InTimeAccumulateWeightedAggregator'

AccumulateWeightedAggregator = 'AccumulateWeightedAggregator'

class flip_api.domain.interfaces.fl.AggregationWeights

MinimumAggregationWeight = 0

MaximumAggregationWeight = 1

flip_api.domain.interfaces.fl.JobTypes

class flip_api.domain.interfaces.fl.JobRequiredFiles(**data)

Bases: pydantic.BaseModel

model_config

classmethod get_required_files(job_type: JobTypes) → list[str]

Returns the list of required files for a specific job type (always reloads from disk).

Parameters:: job_type (JobTypes) – The job type to look up.
Returns:: Required file names for job_type, or an empty list if the job type is not present in the on-disk configuration.
Return type:: list[str]

classmethod get_all_job_types_with_files() → dict[str, list[str]]

Returns all job types with their required files (always reloads from disk).

Returns:: A copy of the on-disk mapping of job type names to their required files.
Return type:: dict[str, list[str]]

classmethod get_job_type_names() → list[str]

Returns a list of all valid job type names.

Returns:: List of job type names.
Return type:: list[str]