Module csw.LocationService

Functions

def csw_version()
Expand source code 
def csw_version():
    return {"csw-version": "6.0.0"}

Classes

class ComponentId (prefix: str, componentType: str)
Expand source code 
@dataclass_json
@dataclass
class ComponentId:
    prefix: str
    componentType: str

ComponentId(prefix: str, componentType: str)

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A
def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A
def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Instance variables

var componentType : str

The type of the None singleton.

var prefix : str

The type of the None singleton.

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]
Expand source code 
def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)
def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str
Expand source code 
def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)
class ComponentType (*args, **kwds)
Expand source code 
class ComponentType(Enum):
    """
    Enum type: Represents a type of the CSW Component.
    """
    Assembly = "Assembly"
    HCD = "HCD"
    Container = "Container"
    Service = "Service"
    Sequencer = "Sequencer"
    SequenceComponent = "SequenceComponent"
    Machine = "Machine"

Enum type: Represents a type of the CSW Component.

Ancestors

  • enum.Enum

Class variables

var Assembly

The type of the None singleton.

var Container

The type of the None singleton.

var HCD

The type of the None singleton.

var Machine

The type of the None singleton.

var SequenceComponent

The type of the None singleton.

var Sequencer

The type of the None singleton.

var Service

The type of the None singleton.

class ConnectionInfo (prefix: str, componentType: str, connectionType: str)
Expand source code 
@dataclass_json
@dataclass
class ConnectionInfo:
    prefix: str
    componentType: str
    connectionType: str

    @staticmethod
    def make(prefix: Prefix, componentType: ComponentType, connectionType: ConnectionType):
        return ConnectionInfo(str(prefix), componentType.value, connectionType.value)

ConnectionInfo(prefix: str, componentType: str, connectionType: str)

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A
def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A
def make(prefix: Prefix,
componentType: ComponentType,
connectionType: ConnectionType)
Expand source code 
@staticmethod
def make(prefix: Prefix, componentType: ComponentType, connectionType: ConnectionType):
    return ConnectionInfo(str(prefix), componentType.value, connectionType.value)
def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Instance variables

var componentType : str

The type of the None singleton.

var connectionType : str

The type of the None singleton.

var prefix : str

The type of the None singleton.

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]
Expand source code 
def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)
def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str
Expand source code 
def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)
class ConnectionType (*args, **kwds)
Expand source code 
class ConnectionType(Enum):
    """
    Enum type: Represents a type of connection offered by the Component.
    Note that PekkoType is for Pekko Actor based connections.
    Python applications can only communicate via HTTP or TCP connections.
    """
    HttpType = "http"
    TcpType = "tcp"
    PekkoType = "pekko"

Enum type: Represents a type of connection offered by the Component. Note that PekkoType is for Pekko Actor based connections. Python applications can only communicate via HTTP or TCP connections.

Ancestors

  • enum.Enum

Class variables

var HttpType

The type of the None singleton.

var PekkoType

The type of the None singleton.

var TcpType

The type of the None singleton.

class HttpLocation (_type: str,
connection: ConnectionInfo,
uri: str,
metadata: dict)
Expand source code 
@dataclass_json
@dataclass
class HttpLocation(Location):
    pass

HttpLocation(_type: str, connection: csw.LocationService.ConnectionInfo, uri: str, metadata: dict)

Ancestors

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A
def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A
def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]
Expand source code 
def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)
def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str
Expand source code 
def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

Inherited members

class HttpRegistration (connection: ConnectionInfo,
port: int,
path: str = '',
networkType: NetworkType = <factory>,
metadata: dict = <factory>)
Expand source code 
@dataclass_json
@dataclass
class HttpRegistration(Registration):
    """
    Used to register an http based service with the Location Service.
    """
    port: int
    path: str = ""
    networkType: NetworkType = field(default_factory=lambda: NetworkType("Inside"))
    metadata: dict = field(default_factory=csw_version)
    _type: str = "HttpRegistration"

Used to register an http based service with the Location Service.

Ancestors

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A
def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A
def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Instance variables

var metadata : dict

The type of the None singleton.

var networkType : NetworkType

The type of the None singleton.

var path : str

The type of the None singleton.

var port : int

The type of the None singleton.

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]
Expand source code 
def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)
def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str
Expand source code 
def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

Inherited members

class Location (_type: str,
connection: ConnectionInfo,
uri: str,
metadata: dict)
Expand source code 
@dataclass_json
@dataclass
class Location:
    _type: str
    connection: ConnectionInfo
    uri: str
    metadata: dict

    @staticmethod
    def _makeLocation(obj: dict):
        match obj['_type']:
            case "HttpLocation":
                return HttpLocation.from_dict(obj)
            case "TcpLocation":
                return TcpLocation.from_dict(obj)
            case "PekkoLocation":
                return PekkoLocation.from_dict(obj)
            case _:
                raise Exception(f"Invalid location type: {obj['_type']}")

Location(_type: str, connection: csw.LocationService.ConnectionInfo, uri: str, metadata: dict)

Subclasses

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A
def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A
def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Instance variables

var connection : ConnectionInfo

The type of the None singleton.

var metadata : dict

The type of the None singleton.

var uri : str

The type of the None singleton.

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]
Expand source code 
def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)
def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str
Expand source code 
def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)
class LocationService (clientSession: aiohttp.client.ClientSession)
Expand source code 
class LocationService:
    log = structlog.get_logger()
    baseUri = "http://127.0.0.1:7654/"
    postUri = f"{baseUri}post-endpoint"

    def __init__(self, clientSession: ClientSession):
        self._session = clientSession

    async def register(self, registration: Registration) -> RegistrationResult:
        """
        Registers a connection.

        Args:
            registration (Registration): the Registration holding the connection and it's corresponding location to
            register with LocationService

        Returns: ConnectionInfo
            an object describing the connection
        """
        # noinspection PyUnresolvedReferences
        regJson = json.loads(registration.to_json())
        regJson['_type'] = registration.__class__.__name__
        jsonBody = f'{{"_type": "Register", "registration": {json.dumps(regJson)}}}'
        r = await self._session.post(self.postUri, json=json.loads(jsonBody))
        if not r.ok:
            raise Exception(r.text)
        maybeResult = json.loads(await r.text())
        if len(maybeResult) != 0:
            location = Location._makeLocation(maybeResult)
            return RegistrationResult.make(location, self.unregister)

    async def unregister(self, connection: ConnectionInfo):
        """
        Unregisters a connection.

        Args:
            connection (ConnectionInfo): an already registered connection
        """
        self.log.debug(f"Unregistering connection {connection} from the Location Service.")
        # noinspection PyUnresolvedReferences
        jsonBody = f'{{"_type": "Unregister", "connection": {connection.to_json()}}}'
        r = await self._session.post(self.postUri, json=json.loads(jsonBody))
        if not r.ok:
            raise Exception(await r.text())

    async def _postJson(self, jsonBody: str) -> Location:
        r = await self._session.post(self.postUri, json=json.loads(jsonBody))
        if not r.ok:
            raise Exception(r.text)
        maybeResult = json.loads(await r.text())
        if len(maybeResult) != 0:
            return Location._makeLocation(maybeResult[0])

    async def find(self, connection: ConnectionInfo) -> Location:
        """
        Resolves the location for a connection from the local cache

        Args:
            connection (ConnectionInfo): an already registered connection

        Returns: Location
            the Location
        """
        # noinspection PyUnresolvedReferences
        return await self._postJson(f'{{"_type": "Find", "connection": {connection.to_json()}}}')

    # "within":"2 seconds"}}
    async def resolve(self, connection: ConnectionInfo, withinSecs: int = "5") -> Location:
        """
        Resolves the location for a connection from the local cache, if not found waits for the event to arrive
        within specified time limit

        Args:
            connection (ConnectionInfo): an already registered connection
            withinSecs (int): max number of seconds to wait for the connection to be found

        Returns: Location
            the Location
        """
        resolveInfo = ResolveInfo("Resolve", connection, f"{withinSecs} seconds")
        # noinspection PyUnresolvedReferences
        return await self._postJson(resolveInfo.to_json())

    async def _list(self, jsonBody: str) -> List[Location]:
        r = await self._session.post(LocationService.postUri, json=json.loads(jsonBody))
        if not r.ok:
            raise Exception(r.text)
        return list(map(lambda x: Location._makeLocation(x), json.loads(await r.text())))

    @dispatch()
    async def list(self) -> List[Location]:
        """
        Lists all locations registered.

        Returns: List[Location]
            list of locations
        """
        jsonBody = '{"_type": "ListEntries"}'
        return await self._list(jsonBody)

    @dispatch(ComponentType)
    async def list(self, componentType: ComponentType) -> List[Location]:
        """
        Lists components of the given component type

        Args:
            componentType (ComponentType): the type of component to list

        Returns: List[Location]
            list of locations
        """
        jsonBody = f'{{"_type": "ListByComponentType", "componentType": "{componentType.value}"}}'
        return await self._list(jsonBody)

    @dispatch(str)
    async def list(self, hostname: str) -> List[Location]:
        """
        Lists all locations registered on the given hostname

        Args:
            hostname (str): restrict result to this host

        Returns: List[Location]
            list of locations
        """
        jsonBody = f'{{"_type": "ListByHostname", "hostname": "{hostname}"}}'
        return await self._list(jsonBody)

    @dispatch(ConnectionType)
    async def list(self, connectionType: ConnectionType) -> List[Location]:
        """
        Lists all locations registered with the given connection type

        Args:
            connectionType (ConnectionType): restrict result to this connection type

        Returns: List[Location]
            list of locations
        """
        jsonBody = f'{{"_type": "ListByConnectionType", "connectionType": "{connectionType.value}"}}'
        return await self._list(jsonBody)

    async def listByPrefix(self, prefix: str) -> List[Location]:
        """
        Lists all locations with the given prefix

        Args:
            prefix (str): restrict result to those starting with prefix

        Returns: List[Location]
            list of locations
        """
        jsonBody = f'{{"_type": "ListByPrefix", "prefix": "{prefix}"}}'
        return await self._list(jsonBody)

Class variables

var baseUri

The type of the None singleton.

var log

The type of the None singleton.

var postUri

The type of the None singleton.

Methods

async def find(self,
connection: ConnectionInfo) ‑> Location
Expand source code 
async def find(self, connection: ConnectionInfo) -> Location:
    """
    Resolves the location for a connection from the local cache

    Args:
        connection (ConnectionInfo): an already registered connection

    Returns: Location
        the Location
    """
    # noinspection PyUnresolvedReferences
    return await self._postJson(f'{{"_type": "Find", "connection": {connection.to_json()}}}')

Resolves the location for a connection from the local cache

Args

connection : ConnectionInfo
an already registered connection

Returns: Location the Location

def list(...)
Expand source code 
def __get__(self, instance, owner):
    self.obj = instance
    self.cls = owner
    return self

Dispatch methods based on type signature

See Also: Dispatcher

async def listByPrefix(self, prefix: str) ‑> List[Location]
Expand source code 
async def listByPrefix(self, prefix: str) -> List[Location]:
    """
    Lists all locations with the given prefix

    Args:
        prefix (str): restrict result to those starting with prefix

    Returns: List[Location]
        list of locations
    """
    jsonBody = f'{{"_type": "ListByPrefix", "prefix": "{prefix}"}}'
    return await self._list(jsonBody)

Lists all locations with the given prefix

Args

prefix : str
restrict result to those starting with prefix

Returns: List[Location] list of locations

async def register(self,
registration: Registration) ‑> RegistrationResult
Expand source code 
async def register(self, registration: Registration) -> RegistrationResult:
    """
    Registers a connection.

    Args:
        registration (Registration): the Registration holding the connection and it's corresponding location to
        register with LocationService

    Returns: ConnectionInfo
        an object describing the connection
    """
    # noinspection PyUnresolvedReferences
    regJson = json.loads(registration.to_json())
    regJson['_type'] = registration.__class__.__name__
    jsonBody = f'{{"_type": "Register", "registration": {json.dumps(regJson)}}}'
    r = await self._session.post(self.postUri, json=json.loads(jsonBody))
    if not r.ok:
        raise Exception(r.text)
    maybeResult = json.loads(await r.text())
    if len(maybeResult) != 0:
        location = Location._makeLocation(maybeResult)
        return RegistrationResult.make(location, self.unregister)

Registers a connection.

Args

registration : Registration
the Registration holding the connection and it's corresponding location to

register with LocationService Returns: ConnectionInfo an object describing the connection

async def resolve(self,
connection: ConnectionInfo,
withinSecs: int = '5') ‑> Location
Expand source code 
async def resolve(self, connection: ConnectionInfo, withinSecs: int = "5") -> Location:
    """
    Resolves the location for a connection from the local cache, if not found waits for the event to arrive
    within specified time limit

    Args:
        connection (ConnectionInfo): an already registered connection
        withinSecs (int): max number of seconds to wait for the connection to be found

    Returns: Location
        the Location
    """
    resolveInfo = ResolveInfo("Resolve", connection, f"{withinSecs} seconds")
    # noinspection PyUnresolvedReferences
    return await self._postJson(resolveInfo.to_json())

Resolves the location for a connection from the local cache, if not found waits for the event to arrive within specified time limit

Args

connection : ConnectionInfo
an already registered connection
withinSecs : int
max number of seconds to wait for the connection to be found

Returns: Location the Location

async def unregister(self,
connection: ConnectionInfo)
Expand source code 
async def unregister(self, connection: ConnectionInfo):
    """
    Unregisters a connection.

    Args:
        connection (ConnectionInfo): an already registered connection
    """
    self.log.debug(f"Unregistering connection {connection} from the Location Service.")
    # noinspection PyUnresolvedReferences
    jsonBody = f'{{"_type": "Unregister", "connection": {connection.to_json()}}}'
    r = await self._session.post(self.postUri, json=json.loads(jsonBody))
    if not r.ok:
        raise Exception(await r.text())

Unregisters a connection.

Args

connection : ConnectionInfo
an already registered connection
class LocationServiceUtil
Expand source code 
class LocationServiceUtil:
    # If port is 0, return a random free port, otherwise the given port
    @staticmethod
    def getFreePort(port: int = 0) -> int:
        if port != 0:
            return port
        with socket() as s:
            s.bind(('', 0))
            return s.getsockname()[1]

Static methods

def getFreePort(port: int = 0) ‑> int
Expand source code 
@staticmethod
def getFreePort(port: int = 0) -> int:
    if port != 0:
        return port
    with socket() as s:
        s.bind(('', 0))
        return s.getsockname()[1]
class NetworkType (_type: str)
Expand source code 
@dataclass_json
@dataclass
class NetworkType:
    """
    NetworkType enum {Outside, Inside)
    """
    _type: str

NetworkType enum {Outside, Inside)

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A
def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A
def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]
Expand source code 
def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)
def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str
Expand source code 
def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)
class PekkoLocation (_type: str,
connection: ConnectionInfo,
uri: str,
metadata: dict)
Expand source code 
@dataclass_json
@dataclass
class PekkoLocation(Location):
    pass

PekkoLocation(_type: str, connection: csw.LocationService.ConnectionInfo, uri: str, metadata: dict)

Ancestors

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A
def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A
def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]
Expand source code 
def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)
def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str
Expand source code 
def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

Inherited members

class PekkoRegistration (connection: ConnectionInfo,
actorRefURI: str,
metadata: dict = <factory>)
Expand source code 
@dataclass_json
@dataclass
class PekkoRegistration(Registration):
    """
    Used to register an http based service with the Location Service.
    """
    actorRefURI: str
    metadata: dict = field(default_factory=dict)
    _type: str = "PekkoRegistration"

Used to register an http based service with the Location Service.

Ancestors

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A
def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A
def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Instance variables

var actorRefURI : str

The type of the None singleton.

var metadata : dict

The type of the None singleton.

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]
Expand source code 
def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)
def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str
Expand source code 
def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

Inherited members

class Registration (connection: ConnectionInfo)
Expand source code 
@dataclass_json
@dataclass
class Registration:
    """
    Abstract base class for registering with the location service
    """
    connection: ConnectionInfo

Abstract base class for registering with the location service

Subclasses

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A
def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A
def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Instance variables

var connection : ConnectionInfo

The type of the None singleton.

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]
Expand source code 
def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)
def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str
Expand source code 
def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)
class RegistrationResult (unregister: Callable,
location: Location)
Expand source code 
@dataclass
class RegistrationResult:
    unregister: Callable
    location: Location

    @classmethod
    def make(cls, _location: Location, _unregister: Callable[[ConnectionInfo], Awaitable]) -> Self:
        return cls(lambda: _unregister(_location.connection), _location)

RegistrationResult(unregister: Callable, location: csw.LocationService.Location)

Static methods

def make(_location: Location,
_unregister: Callable[[ConnectionInfo], ]) ‑> Self

Instance variables

var location : Location

The type of the None singleton.

var unregister : Callable

The type of the None singleton.

class ResolveInfo (_type: str,
connection: ConnectionInfo,
within: str)
Expand source code 
@dataclass_json
@dataclass
class ResolveInfo:
    _type: str
    connection: ConnectionInfo
    within: str

ResolveInfo(_type: str, connection: csw.LocationService.ConnectionInfo, within: str)

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A
def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A
def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Instance variables

var connection : ConnectionInfo

The type of the None singleton.

var within : str

The type of the None singleton.

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]
Expand source code 
def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)
def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str
Expand source code 
def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)
class TcpLocation (_type: str,
connection: ConnectionInfo,
uri: str,
metadata: dict)
Expand source code 
@dataclass_json
@dataclass
class TcpLocation(Location):
    pass

TcpLocation(_type: str, connection: csw.LocationService.ConnectionInfo, uri: str, metadata: dict)

Ancestors

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A
def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A
def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]
Expand source code 
def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)
def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str
Expand source code 
def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

Inherited members

class TcpRegistration (connection: ConnectionInfo,
port: int,
metadata: dict = <factory>)
Expand source code 
@dataclass_json
@dataclass
class TcpRegistration(Registration):
    """
    Used to register a tcp based service with the Location Service.
    """
    port: int
    metadata: dict = field(default_factory=dict)
    _type: str = "TcpRegistration"

Used to register a tcp based service with the Location Service.

Ancestors

Static methods

def from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) ‑> ~A
def from_json(s: str | bytes | bytearray,
*,
parse_float=None,
parse_int=None,
parse_constant=None,
infer_missing=False,
**kw) ‑> ~A
def schema(*,
infer_missing: bool = False,
only=None,
exclude=(),
many: bool = False,
context=None,
load_only=(),
dump_only=(),
partial: bool = False,
unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Instance variables

var metadata : dict

The type of the None singleton.

var port : int

The type of the None singleton.

Methods

def to_dict(self, encode_json=False) ‑> Dict[str, dict | list | str | int | float | bool | None]
Expand source code 
def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)
def to_json(self,
*,
skipkeys: bool = False,
ensure_ascii: bool = True,
check_circular: bool = True,
allow_nan: bool = True,
indent: int | str | None = None,
separators: Tuple[str, str] | None = None,
default: Callable | None = None,
sort_keys: bool = False,
**kw) ‑> str
Expand source code 
def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Optional[Tuple[str, str]] = None,
            default: Optional[Callable] = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

Inherited members