epicsarchiver_mgmt.archiver.base

Base Archiver HTTP client module.

Attributes

LOG

Exceptions

ArchiverError	Base class for all exceptions raised by the archiver HTTP client.
ArchiverConnectionError	Exception raised when there is a connection error with the archiver.
ArchiverResponseError	Exception raised when the archiver returns an unexpected response.

Classes

BaseArchiverAppliance

Base EPICS Archiver Appliance client.

Functions

mgmt_url(→ str)

Generate the mgmt url from a hostname and a port number.

Module Contents

epicsarchiver_mgmt.archiver.base.LOG: logging.Logger

epicsarchiver_mgmt.archiver.base.mgmt_url(hostname: str, port: int) → str

Generate the mgmt url from a hostname and a port number.

Parameters:

• hostname (str) – fqdn of service

• port (int) – Port number

Returns:

Completed url, for example “http://localhost:17665/mgmt/bpl/”

Return type:

str

exception epicsarchiver_mgmt.archiver.base.ArchiverError

Bases: epicsarchiver_mgmt.exceptions.BaseMgmtError

Base class for all exceptions raised by the archiver HTTP client.

exception epicsarchiver_mgmt.archiver.base.ArchiverConnectionError(base_url: str, message: str | None = None)

Bases: ArchiverError

Exception raised when there is a connection error with the archiver.

base_url

exception epicsarchiver_mgmt.archiver.base.ArchiverResponseError(base_url: str, url: str | None = None, response: str | None = None, message: str | None = None)

Bases: ArchiverError

Exception raised when the archiver returns an unexpected response.

base_url

url = None

response = None

class epicsarchiver_mgmt.archiver.base.BaseArchiverAppliance(hostname: str = 'localhost', port: int = 17665)

Base EPICS Archiver Appliance client.

Hold a session to the Archiver Appliance web application.

Parameters:

• hostname – EPICS Archiver Appliance hostname [default: localhost]

• port – EPICS Archiver Appliance management port [default: 17665]

hostname = 'localhost'

port = 17665

mgmt_url = 'http://localhost:17665/mgmt/bpl/'

_info: dict[str, str]

_data_retrieval_url: str | None = None

session

__repr__() → str

String representation of Archiver Appliance.

Returns:

details including hostname of Archiver appliance.

Return type:

str

_request(method: str, url: str, params: dict[str, str | list[str]] | dict[str, str] | None = None, data: str | dict[str, str] | None = None, json_data: Any | None = None) → requests.Response

Send a request using the session.

Parameters:

• method – HTTP method

• url – The URL to send the request to

• params – Optional arguments

• data – Optional data to be sent

• json_data – Optional JSON data to be sent

Returns:

requests.Response object

Raises:

• ArchiverConnectionError – If there is a connection error.

• ArchiverResponseError – If the response is not successful.

_get(endpoint: str, params: dict[str, str | list[str]] | dict[str, str] | None = None) → requests.Response

Send a GET request to the given endpoint.

Parameters:

• endpoint – API endpoint (relative or absolute)

• params – Optional arguments to be sent

Returns:

requests.Response object

_post(endpoint: str, params: dict[str, str | list[str]] | dict[str, str] | None = None, data: str | dict[str, str] | None = None, json_data: Any | None = None) → requests.Response

Send a POST request to the given endpoint.

Parameters:

• endpoint – API endpoint (relative or absolute)

• params – Optional arguments to be sent

• data – Optional data to be sent

• json_data – Optional JSON data to be sent

Returns:

requests.Response object

property info: dict[str, str]

EPICS Archiver Appliance information.

property identity: str | None

EPICS Archiver Appliance identity.

property version: str | None

EPICS Archiver Appliance version.

_get_or_post(endpoint: str, pv: str) → Any

Send a GET or POST if pv is a comma separated list.

Parameters:

• endpoint (str) – API endpoint

• pv (str) – name of the pv. Can be a GLOB wildcards or a list of comma separated names.

Returns:

list of submitted PVs

Return type:

Any