fedn.network.api package

API module for the FEDn network. Includes a REST-API server to interact with the controller and statestore.

Submodules

fedn.network.api.client module

class fedn.network.api.client.APIClient(host, port=None, secure=False, verify=False, token=None, auth_scheme=None)[source]

Bases: object

An API client for interacting with the statestore and controller.

Parameters:

host (str) – The host of the api server.
port (int) – The port of the api server.
secure (bool) – Whether to use https.
verify (bool) – Whether to verify the server certificate.

download_model(id: str, path: str)[source]

Download the model with id id.

Parameters:

id (str) – The id (or model property) of the model to download.
path (str) – The path to download the model to.

Returns:

Message with success or failure.

Return type:

dict

download_package(path: str)[source]

Download the compute package.

Parameters:: path (str) – The path to download the compute package to.
Returns:: Message with success or failure.
Return type:: dict

get_active_clients(combiner_id: str | None = None, n_max: int | None = None)[source]

Get active clients from the statestore.

Parameters:

combiner_id (str) – The combiner id to get active clients for.
n_max (int) – The maximum number of clients to get (If none all will be fetched).

Returns:

Active clients.

Return type:

dict

get_active_model()[source]

Get the latest model from the statestore.

Returns:: The latest model.
Return type:: dict

get_active_package()[source]

Get the (active) compute package from the statestore.

Returns:: Package.
Return type:: dict

get_client(id: str)[source]

Get a client from the statestore.

Parameters:: id (str) – The client id to get.
Returns:: Client.
Return type:: dict

get_client_config(checksum=True)[source]

Get client config from controller. Optionally include the checksum. The config is used for clients to connect to the controller and ask for combiner assignment.

Parameters:: checksum (bool) – Whether to include the checksum of the package.
Returns:: The client configuration.
Return type:: dict

get_clients(n_max: int | None = None)[source]

Get clients from the statestore.

Parameters:: n_max (int) – The maximum number of clients to get (If none all will be fetched).

return: Clients. rtype: dict

get_clients_count()[source]

Get the number of clients in the statestore.

Returns:: The number of clients.
Return type:: dict

get_combiner(id: str)[source]

Get a combiner from the statestore.

Parameters:: id (str) – The combiner id to get.
Returns:: Combiner.
Return type:: dict

get_combiners(n_max: int | None = None)[source]

Get combiners in the network.

Parameters:: n_max (int) – The maximum number of combiners to get (If none all will be fetched).
Returns:: Combiners.
Return type:: dict

get_combiners_count()[source]

Get the number of combiners in the statestore.

Returns:: The number of combiners.
Return type:: dict

get_controller_status()[source]

Get the status of the controller.

Returns:: The status of the controller.
Return type:: dict

get_model(id: str)[source]

Get a model from the statestore.

Parameters:: id (str) – The id (or model property) of the model to get.
Returns:: Model.
Return type:: dict

get_model_trail(id: str | None = None, include_self: bool = True, reverse: bool = True, n_max: int | None = None)[source]

Get the model trail.

Parameters:

id (str) – The id (or model property) of the model to start the trail from. (optional)
n_max (int) – The maximum number of models to get (If none all will be fetched).

Returns:

Models.

Return type:

dict

get_models(session_id: str | None = None, n_max: int | None = None)[source]

Get models from the statestore.

Parameters:

session_id (str) – The session id to get models for. (optional)
n_max (int) – The maximum number of models to get (If none all will be fetched).

Returns:

Models.

Return type:

dict

get_models_count()[source]

Get the number of models in the statestore.

Returns:: The number of models.
Return type:: dict

get_package(id: str)[source]

Get a compute package from the statestore.

Parameters:: id (str) – The id of the compute package to get.
Returns:: Package.
Return type:: dict

get_package_checksum()[source]

Get the checksum of the compute package.

Returns:: The checksum.
Return type:: dict

get_packages(n_max: int | None = None)[source]

Get compute packages from the statestore.

Parameters:: n_max (int) – The maximum number of packages to get (If none all will be fetched).
Returns:: Packages.
Return type:: dict

get_packages_count()[source]

Get the number of compute packages in the statestore.

Returns:: The number of packages.
Return type:: dict

get_round(id: str)[source]

Get a round from the statestore.

Parameters:: round_id (str) – The round id to get.
Returns:: Round (config and metrics).
Return type:: dict

get_rounds(n_max: int | None = None)[source]

Get all rounds from the statestore.

Parameters:: n_max (int) – The maximum number of rounds to get (If none all will be fetched).
Returns:: Rounds.
Return type:: dict

get_rounds_count()[source]

Get the number of rounds in the statestore.

Returns:: The number of rounds.
Return type:: dict

get_session(id: str)[source]

Get a session from the statestore.

Parameters:: id (str) – The session id to get.
Returns:: Session.
Return type:: dict

get_session_status(id: str)[source]

Get the status of a session.

Parameters:: id (str) – The id of the session to get.
Returns:: The status of the session.
Return type:: str

get_sessions(n_max: int | None = None)[source]

Get sessions from the statestore.

Parameters:: n_max (int) – The maximum number of sessions to get (If none all will be fetched).
Returns:: Sessions.
Return type:: dict

get_sessions_count()[source]

Get the number of sessions in the statestore.

Returns:: The number of sessions.
Return type:: dict

get_status(id: str)[source]

Get a status object (event) from the statestore.

Parameters:: id (str) – The id of the status to get.
Returns:: Status.
Return type:: dict

Get statuses from the statestore. Filter by input parameters

Parameters:

session_id (str) – The session id to get statuses for.
event_type (str) – The event type to get.
sender_name (str) – The sender name to get.
sender_role (str) – The sender role to get.
n_max (int) – The maximum number of statuses to get (If none all will be fetched).

Returns:

Statuses

get_statuses_count()[source]

Get the number of statuses in the statestore.

Returns:: The number of statuses.
Return type:: dict

get_validation(id: str)[source]

Get a validation from the statestore.

Parameters:: id (str) – The id of the validation to get.
Returns:: Validation.
Return type:: dict

Get validations from the statestore. Filter by input parameters.

Parameters:

session_id (str) – The session id to get validations for.
model_id (str) – The model id to get validations for.
correlation_id (str) – The correlation id to get validations for.
sender_name (str) – The sender name to get validations for.
sender_role (str) – The sender role to get validations for.
receiver_name (str) – The receiver name to get validations for.
receiver_role (str) – The receiver role to get validations for.
n_max (int) – The maximum number of validations to get (If none all will be fetched).

Returns:

Validations.

Return type:

dict

get_validations_count()[source]

Get the number of validations in the statestore.

Returns:: The number of validations.
Return type:: dict

session_is_finished(id: str)[source]

Check if a session with id has finished.

Parameters:: id (str) – The id of the session to get.
Returns:: True if session is finished, otherwise false.
Return type:: bool

set_active_model(path)[source]

Set the initial model in the statestore and upload to model repository.

Parameters:: path (str) – The file path of the initial model to set.
Returns:: A dict with success or failure message.
Return type:: dict

set_active_package(path: str, helper: str, name: str | None = None, description: str | None = None)[source]

Set the compute package in the statestore.

Parameters:

path (str) – The file path of the compute package to set.
helper (str) – The helper type to use.

Returns:

A dict with success or failure message.

Return type:

dict

start_session(id: str | None = None, aggregator: str = 'fedavg', aggregator_kwargs: dict | None = None, model_id: str | None = None, round_timeout: int = 180, rounds: int = 5, round_buffer_size: int = -1, delete_models: bool = True, validate: bool = True, helper: str = 'numpyhelper', min_clients: int = 1, requested_clients: int = 8)[source]

Start a new session.

Parameters:

id (str) – The session id to start.
aggregator (str) – The aggregator plugin to use.
model_id (str) – The id of the initial model.
round_timeout (int) – The round timeout to use in seconds.
rounds (int) – The number of rounds to perform.
round_buffer_size (int) – The round buffer size to use.
delete_models (bool) – Whether to delete models after each round at combiner (save storage).
validate (bool) – Whether to validate the model after each round.
helper (str) – The helper type to use.
min_clients (int) – The minimum number of clients required.
requested_clients (int) – The requested number of clients.

Returns:

A dict with success or failure message and session config.

Return type:

dict

fedn.network.api.interface module

class fedn.network.api.interface.API(statestore, control)[source]

Bases: object

The API class is a wrapper for the statestore. It is used to expose the statestore to the network API.

add_client(client_id, preferred_combiner, remote_addr)[source]

Add a client to the network.

Parameters:

client_id (str) – The client id to add.
preferred_combiner (str) – The preferred combiner for the client.If None, the combiner will be chosen based on availability.

Returns:

A json response with combiner assignment config.

Return type:

flask.Response

add_combiner(combiner_id, secure_grpc, address, remote_addr, fqdn, port)[source]

Add a combiner to the network.

Parameters:

combiner_id (str) – The combiner id to add.
secure_grpc (bool) – Whether to use secure grpc or not.
name (str) – The name of the combiner.
address (str) – The address of the combiner.
remote_addr (str) – The remote address of the combiner.
fqdn (str) – The fqdn of the combiner.
port (int) – The port of the combiner.

Returns:

Config of the combiner as a json response.

Return type:

flask.Response

download_compute_package(name)[source]

Download the compute package.

Returns:: The compute package as a json object.
Return type:: flask.Response

get_all_combiners(limit=None, skip=None)[source]

Get all combiners from the statestore.

Returns:: All combiners as a json response.
Return type:: flask.Response

get_all_rounds()[source]

Get all rounds.

Returns:: The rounds as json response.
Return type:: flask.Response

get_all_sessions(limit=None, skip=None)[source]

Get all sessions from the statestore.

Returns:: All sessions as a json response.
Return type:: flask.Response

get_all_validations(**kwargs)[source]

Get all validations from the statestore.

Returns:: All validations as a json response.
Return type:: flask.Response

get_checksum(name)[source]

Get the checksum of the compute package.

Parameters:: name (str) – The name of the compute package.
Returns:: The checksum as a json object.
Return type:: flask.Response

get_client_config(checksum=True)[source]

Get the client config.

Returns:: The client config as json response.
Return type:: flask.Response

get_clients(limit=None, skip=None, status=False)[source]

Get all clients from the statestore.

Returns:: All clients as a json response.
Return type:: flask.Response

get_combiner(combiner_id)[source]

Get a combiner from the statestore.

Parameters:: combiner_id (str) – The combiner id to get.
Returns:: The combiner info dict as a json response.
Return type:: flask.Response

get_compute_package()[source]

Get the compute package from the statestore.

Returns:: The compute package as a json response.
Return type:: flask.Response

get_controller_status()[source]

Get the status of the controller.

Returns:: The status of the controller as a json object.
Return type:: flask.Response

get_events(**kwargs)[source]

Get the events of the federated network.

Returns:: The events as a json object.
Return type:: flask.Response

get_initial_model()[source]

Get the initial model from the statestore.

Returns:: The initial model as a json response.
Return type:: flask.Response

get_latest_model()[source]

Get the latest model from the statestore.

Returns:: The initial model as a json response.
Return type:: flask.Response

get_model(model_id: str)[source]

get_model_ancestors(model_id: str, limit: str | None = None)[source]

Get the model ancestors for a given model.

Parameters:

model_id (str) – The model id to get the model ancestors for.
limit (str) – The number of ancestors to return.

Returns:

The model ancestors for the given model as a json response.

Return type:

flask.Response

get_model_descendants(model_id: str, limit: str | None = None)[source]

Get the model descendants for a given model.

Parameters:

model_id (str) – The model id to get the model descendants for.
limit (str) – The number of descendants to return.

Returns:

The model descendants for the given model as a json response.

Return type:

flask.Response

get_model_trail()[source]

Get the model trail for a given session.

Parameters:: session (str) – The session id to get the model trail for.
Returns:: The model trail for the given session as a json response.
Return type:: flask.Response

get_models(session_id: str | None = None, limit: str | None = None, skip: str | None = None, include_active: str | None = None)[source]

get_plot_data(feature=None)[source]

Get plot data.

Returns:: The plot data as json response.
Return type:: flask.Response

get_round(round_id)[source]

Get a round.

Parameters:: round_id (str) – The round id to get.
Returns:: The round as json response.
Return type:: flask.Response

get_session(session_id)[source]

Get a session from the statestore.

Parameters:: session_id (str) – The session id to get.
Returns:: The session info dict as a json response.
Return type:: flask.Response

list_combiners_data(combiners)[source]

Get combiners data.

Parameters:: combiners (list) – The combiners to get data for.
Returns:: The combiners data as json response.
Return type:: flask.Response

list_compute_packages(limit: str | None = None, skip: str | None = None, include_active: str | None = None)[source]: Get paginated list of compute packages from the statestore. :param limit: The number of compute packages to return. :type limit: str :param skip: The number of compute packages to skip. :type skip: str :param include_active: Whether to include the active compute package or not. :type include_active: str :return: All compute packages as a json response. :rtype: flask.Response

set_active_compute_package(id: str)[source]

set_compute_package(file, helper_type: str, name: str | None = None, description: str | None = None)[source]

Set the compute package in the statestore.

Parameters:: file (file) – The compute package to set.
Returns:: A json response with success or failure message.
Return type:: flask.Response

set_current_model(model_id: str)[source]

Set the active model in the statestore.

Parameters:: model_id (str) – The model id to set.
Returns:: A json response with success or failure message.
Return type:: flask.Response

set_initial_model(file)[source]

Add an initial model to the network.

Parameters:: file (file) – The initial model to add.
Returns:: A json response with success or failure message.
Return type:: flask.Response

start_session(session_id, aggregator='fedavg', aggregator_kwargs=None, model_id=None, rounds=5, round_timeout=180, round_buffer_size=-1, delete_models=True, validate=True, helper='numpyhelper', min_clients=1, requested_clients=8)[source]

Start a session.

Parameters:

session_id (str) – The session id to start.
aggregator (str) – The aggregator plugin to use.
initial_model (str) – The initial model for the session.
rounds (int) – The number of rounds to perform.
round_timeout (int) – The round timeout to use in seconds.
round_buffer_size (int) – The round buffer size to use.
delete_models (bool) – Whether to delete models after each round at combiner (save storage).
validate (bool) – Whether to validate the model after each round.
min_clients (int) – The minimum number of clients required.
requested_clients (int) – The requested number of clients.

Returns:

A json response with success or failure message and session config.

Return type:

flask.Response

fedn.network.api.network module

class fedn.network.api.network.Network(control, statestore, load_balancer=None)[source]

Bases: object

FEDn network interface. This class is used to interact with the network. Note: This class contain redundant code, which is not used in the current version of FEDn. Some methods has been moved to fedn.network.api.interface.API.

add_client(client)[source]

Add a new client to the network.

Parameters:: client (dict) – The client instance object
Returns:: None

add_combiner(combiner)[source]

Add a new combiner to the network.

Parameters:: combiner (fedn.network.combiner.interfaces.CombinerInterface) – The combiner instance object
Returns:: None

find_available_combiner()[source]

Find an available combiner in the network.

Returns:: The combiner instance object
Return type:: fedn.network.combiner.interfaces.CombinerInterface

get_client(name)[source]

Get client by name.

Parameters:: name (str) – name of client
Returns:: The client instance object
Return type:: ObjectId

get_client_info()[source]

List available client in statestore.

Returns:: list of client objects
Return type:: list(ObjectId)

get_combiner(name)[source]

Get combiner by name.

Parameters:: name (str) – name of combiner
Returns:: The combiner instance object
Return type:: fedn.network.combiner.interfaces.CombinerInterface

get_combiners()[source]

Get all combiners in the network.

Returns:: list of combiners objects
Return type:: list(fedn.network.combiner.interfaces.CombinerInterface)

handle_unavailable_combiner(combiner)[source]

This callback is triggered if a combiner is found to be unresponsive.

Parameters:: combiner (fedn.network.combiner.interfaces.CombinerInterface) – The combiner instance object
Returns:: None

remove_combiner(combiner)[source]

Remove a combiner from the network.

Parameters:: combiner (fedn.network.combiner.interfaces.CombinerInterface) – The combiner instance object
Returns:: None

update_client_data(client_data, status, role)[source]

Update client status in statestore.

Parameters:

client_data (dict) – The client instance object
status (str) – The client status
role (str) – The client role

Returns:

None

fedn.network.api.tests module

class fedn.network.api.tests.NetworkAPITests(methodName='runTest')[source]

Bases: TestCase

Unittests for the Network API.

setUp(mock_mongo, mock_control)[source]: Hook method for setting up the test fixture before exercising it.

test_add_client()[source]: Test add_client endpoint.

test_add_combiner()[source]: Test add_combiner endpoint.

test_get_active_clients()[source]: Test get_active_clients endpoint.

test_get_client_config()[source]: Test get_client_config endpoint.

test_get_combiner()[source]: Test get_combiner endpoint.

test_get_controller_status()[source]: Test get_controller_status endpoint.

test_get_events()[source]: Test get_events endpoint.

test_get_initial_model()[source]: Test get_initial_model endpoint.

test_get_latest_model()[source]: Test get_latest_model endpoint.

test_get_model_trail()[source]: Test get_model_trail endpoint.

test_get_package()[source]: Test get_package endpoint.

test_get_round()[source]: Test get_round endpoint.

test_get_status()[source]: Test get_status endpoint.

test_list_clients()[source]: Test list_clients endpoint.

test_list_combiners()[source]: Test list_combiners endpoint.

test_list_compute_packages()[source]: Test list_compute_packages endpoint.

test_list_models()[source]: Test list_models endpoint.

test_list_rounds()[source]: Test list_rounds endpoint.

test_list_sessions()[source]: Test list_sessions endpoint.

test_set_initial_model()[source]: Test set_initial_model endpoint.

test_start_session()[source]: Test start_session endpoint.