ipcs package

Submodules

ipcs.client module

class ipcs.client.Request(source: Connection, session: str, route: str, raw: RequestPayload)

Bases: object

Data class for storing request information.

source: Connection

The sender who made the request.

session: str

str ID to identify the request.

route: str

The name of what str of execution should be executed in the request.

raw: RequestPayload

This is the raw data of the request.

classmethod from_payload(source: Connection, data: RequestPayload) → Request

Create an instance of this class from the requestor connection and raw data.

class ipcs.client.AbcClient(id_: str, timeout: float = 8.0)

Bases: ABC, Generic[ConnectionT]

Base class for the client class.

Parameters:

  • id – Client ID. It must be unique among clients connecting to the server.

  • timeout – When a client makes a request, how long it waits for that request.

routes

A dictionary for storing registered strs.

listeners

A dictionary to store listeners for registered events.

connections

A dictionary to store ipcs.connection.Connection of clients connecting to the server. This dictionary uses ipcs.utils.SimpleAttrDict and allows access to dictionary values from attributes.

id_

Client ID.

timeout

When a client makes a request, how long it waits for that request.

ws: WebSocketProtocol | None = None

The web socket used by the client to communicate with the server.

property loop: AbstractEventLoop

The event loop used by the client.

add_listener(func: Callable[[...], Any | Coroutine[Any, Any, Any]], name: str | None = None) → None

Add a function (called a listener) that executes the event when it occurs. The default available events can be found in the EventReference menu.

Parameters:

  • func – Function to be executed when an event occurs.

  • name – The name of what event to listen for. If None, the function name is used instead.

remove_listener(target: Callable[[...], Any | Coroutine[Any, Any, Any]] | str) → None

The name of what event to listen for. If None, the function name is used instead.

Parameters:

target – The name or function of the listener to be deleted.

listen(name: str | None = None) → Callable[[LiT], LiT]

Decorator version of add_listener().

Parameters:

name – The name of what event to listen for. If None, the function name is used instead.

dispatch(name: str, *args: Any, **kwargs: Any) → None

Raises an event.

Parameters:

  • name – The name of the event.

  • *args – The arguments to be passed to the event handler (listener).

  • **kwargs – The keyword arguments to be passed to the event handler (listener).

set_route(func: Callable[[...], Any | Coroutine[Any, Any, Any]], name: str | None = None, secret: bool = False) → None

Registers a function to respond to a request from another client. Now the registered function can be called by another client with the name used for registration. (This is called a request.) Such a function is also called a str.

Parameters:

  • func – Function to register as str.

  • name – str Name. If None, the function name is used.

  • secret – If it is True, function is to be registered as a hidden str. This is used internally in ipcs to add a str to receive information when a client connects from the server to a new connection, for example. A str registered in this way cannot be deleted. Therefore, it should not be used outside of ipcs.

remove_route(target: Callable[[...], Any | Coroutine[Any, Any, Any]] | str) → None

Delete a str.

Parameters:

target – Name or function of str to be deleted.

route(name: str | None = None, secret: bool = False) → Callable[[RhReT], RhReT]

Decorator version of set_route().

Parameters:
generate_session() → str

Generates a session ID used to identify the request. This is used inside ipcs.

async request(id_: str, route: str, *args: Any, **kwargs: Any) → Any

Sends a request to the specified connection. Alias of ipcs.connection.Connection.request().

Parameters:

  • id – The id of the connection.

  • route – The route to request.

  • *args – The arguments to be passed to str.

  • **kwargs – The keyword arguments to be passed to str.

Raises:
async request_all(route: str, *args: ~typing.Any, key_: ~collections.abc.Callable[[~ipcs.connection.Connection], bool] = <function AbcClient.<lambda>>, **kwargs: ~typing.Any) → dict[str, tuple[Optional[Any], None | ipcs.errors.RequestError]]

Make requests to everyone except yourself.

Parameters:

  • route – The name of the str to request.

  • *args – The arguments to be passed to str.

  • key – Function executed to determine if a request should be sent. This can be used, for example, to send a request only to IDs that conform to certain rules.

  • **kwargs – The keyword arguments to be passed to str.

Returns:

The return value is a dictionary whose key is the ID of the requestor and whose value is a tuple containing the return value and the error. If no error occurred, the error value is None.

abstract async start(*args: Any, **kwargs: Any) → None

This method is not fully implemented. Implemented are ipcs.client.Client and ipcs.server.Server.

abstract async close(code: int = 1000, reason: str = '...') → None

This method is not fully implemented. Implemented are ipcs.client.Client and ipcs.server.Server.

run(*args: Any, **kwargs: Any) → None

Start client. It internally uses run in the standard library asyncio to run ipcs.AbcClient.start().

Parameters:
class ipcs.client.Client(*args: Any, **kwargs: Any)

Bases: AbcClient[Connection]

This is the class of the client for ipc communication with the server.

ready: Event

This is an Event in the standard library asyncio to put if it has already started (connected to the server). You can use await ready.wait() to wait until it starts.

async start(*args: Any, reconnect: bool = True, **kwargs: Any) → None

Starts (connects) client.

Parameters:

  • *args – The arguments to be passed to connect in the third party library websockets. Details of connect is here.

  • reconnect – Whether or not to make the reconnection.

  • **kwargs –

    The keyword arguments to be passed to connect in the third party library websockets. Details of connect is here.

async close(code: int = 1000, reason: str = '...') → None

Disconnects the client from the server.

Parameters:

  • code – This code is used when cutting with a websocket.

  • reason – This reason is used when cutting with a websocket.

ipcs.client.logger = <Logger ipcs (WARNING)>

Log output destination. ipcs use logging from the standard library.

ipcs.connection module

class ipcs.connection.Connection(client: AbcClient, id_: str)

Bases: object

This class is used to represent the connection to the client. Requests to the client can be made using this.

Parameters:

  • client – This is the client who established this connection.

  • id – The ID of the client connected to this connection.

client

This is the client who established this connection.

id_

The ID of the client connected to this connection.

async close() → None

Disconnect this connection.

async request(route: str, *args: Any, ipcs_secret: bool = False, **kwargs: Any) → Any

Make a request to the client connected by this connection.

Parameters:

  • route – The name of the str you wish to execute at the request destination.

  • *args – The arguments to be passed to the str.

  • ipcs_secret – Whether the request is for internal use by ipcs. Do not set this to True for normal use.

  • **kwargs – The arguments to be passed to the str.

Raises:

ipcs.errors module

exception ipcs.errors.IpcsError

Bases: Exception

All errors defined in ipcs are inherited errors.

exception ipcs.errors.IdIsNotFound

Bases: IpcsError

This error occurs when str is not found.

exception ipcs.errors.RequestError

Bases: IpcsError

This error occurs when a request fails.

exception ipcs.errors.FailedToRequestError

Bases: RequestError

This error occurs when a request failes at client. What error occurred goes into the attribute __cause__. Example: asyncio.TimeoutError

exception ipcs.errors.FailedToProcessError(message: str, error: str, *args: Any, **kwargs: Any)

Bases: RequestError

Occurs when an error occurs at the request destination.

error

The name and description of the error that occurred at the request destination. The format is <name>: <content>.

exception ipcs.errors.ClosedConnectionError

Bases: RequestError

This occurs when a request is attempted but communication is lost.

ipcs.server module

class ipcs.server.ConnectionForServer(client: AbcClient, id_: str)

Bases: Connection

Class that extends Connection for Server.

Parameters:

ws – This is the websocket to the client connected by this connection.

ws: WebSocketProtocol

This is the websocket to the client connected by this connection.

async close() → None

Disconnect this connection.

class ipcs.server.Server(*args: Any, **kwargs: Any)

Bases: AbcClient[ConnectionForServer | Connection]

Server for ipcs communication. This is not normally used to run ipcs servers. Instead, use ipcs-server, which is available on the command line. This command is automatically set up during installation with pip.

async communicate(ws: WebSocketProtocol) → None

Function called when a new client connects. Used internally. ipcs uses web sockets using a third-party library called websocekts, but if you want to use another web socket library, pass the newly connected web socket to this function. However, since the websockets available to ipcs must conform to ipcs.types_.WebSocketProtocol, you will need to wrap the websocket class to make it conform to ipcs.types_.WebSocketProtocol if it differs from that.

Parameters:

ws – The websocket of the newly connected client.

async start(*args: Any, **kwargs: Any) → None

Run the server.

Parameters:

  • *args –

    The arguments to be passed to serve in the third party library websockets. Details of serve in here.

  • **kwargs –

    The keyword arguments to be passed to serve in the third party library websockets. Details of serve in here.

async close(code: int = 1000, reason: str = '...') → None

Terminate the server.

Parameters:

  • code – This code is used when cutting with a websocket.

  • reason – This reason is used when cutting with a websocket.

ipcs.types_ module

class ipcs.types_.BasePayload

Bases: TypedDict

The base JSON type of the raw data used for communication.

source: str

The ID of the data source.

target: str

The ID of the destination of the data.

secret: bool

It is whether the data is directed to the inside of ipcs or not.

session: str

str ID to identify data.

route: str

str Name.

class ipcs.types_.RequestPayload

Bases: dict

The JSON type of the raw data at the time of the request.

type: Literal['request']

Indicates the type of data.

args: Sequence[Any]

Arguments to be passed to str.

kwargs: dict[str, Any]

Keyword arguments to be passed to str.

source: str
target: str
secret: bool
session: str
route: str
class ipcs.types_.ResponsePayload

Bases: dict

The JSON type of the raw data at response time.

source: str
target: str
secret: bool
session: str
route: str
type: Literal['response']

Indicates the type of data.

status: Literal['ok', 'error']

A string representing the result of the request execution.

result: Any

Stores the results of the request execution. If there is an error, this will contain a string with the name of the error and its contents.

class ipcs.types_.WebSocketProtocol(*args, **kwargs)

Bases: Protocol

Protocol class to indicate functions that must be implemented in the class for WebSocket communication used within ipcs.

property closed: bool

Returns whether the websocket is closed or not.

async send(data: Any, *args: Any, **kwargs: Any) → None

Sends data via websockets.

async recv() → Any

Receives data via websockets.

async close(code: int = 1000, reason: str = '...') → None

Disconnect the web socket.

ipcs.utils module

class ipcs.utils.SimpleAttrDict

Bases: dict[str, ValueT]

This class is a dictionary that allows values to be retrieved from attributes. Note, however, that dictionaries retrieved from attributes cannot be accessed from attributes.

class ipcs.utils.DataEvent(*, loop=<object object>)

Bases: Event, Generic[DataT]

This class extends the Event class of the standard asyncio library to allow setting some object.

data: DataT | None = None
null = False
clear() → None

Reset the internal flag to false. Subsequently, coroutines calling wait() will block until set() is called to set the internal flag to true again.

set_with_null() → None

Runs set() but does not set any data. In this case, wait() raises a ClosedConnectionError.

set(data: DataT) → None

Set the internal flag to true. All coroutines waiting for it to become true are awakened. Coroutine that call wait() once the flag is true will not block at all.

async wait() → DataT

Block until the internal flag is true.

If the internal flag is true on entry, return True immediately. Otherwise, block until another coroutine calls set() to set the flag to true, then return True.

ipcs.utils.payload_to_str(payload: RequestPayload | ResponsePayload) → str

Represents raw data as a string.

Module contents

class ipcs.Request(source: Connection, session: str, route: str, raw: RequestPayload)

Bases: object

Data class for storing request information.

source: Connection

The sender who made the request.

session: str

str ID to identify the request.

route: str

The name of what str of execution should be executed in the request.

raw: RequestPayload

This is the raw data of the request.

classmethod from_payload(source: Connection, data: RequestPayload) → Request

Create an instance of this class from the requestor connection and raw data.

class ipcs.AbcClient(id_: str, timeout: float = 8.0)

Bases: ABC, Generic[ConnectionT]

Base class for the client class.

Parameters:

  • id – Client ID. It must be unique among clients connecting to the server.

  • timeout – When a client makes a request, how long it waits for that request.

routes

A dictionary for storing registered strs.

Type:

dict[str, collections.abc.Callable[…, Union[Any, collections.abc.Coroutine[Any, Any, Any]]]]

listeners

A dictionary to store listeners for registered events.

Type:

collections.defaultdict[str, list[collections.abc.Callable[…, Union[Any, collections.abc.Coroutine[Any, Any, Any]]]]]

connections

A dictionary to store ipcs.connection.Connection of clients connecting to the server. This dictionary uses ipcs.utils.SimpleAttrDict and allows access to dictionary values from attributes.

Type:

ipcs.utils.SimpleAttrDict[ipcs.client.ConnectionT]

id_

Client ID.

timeout

When a client makes a request, how long it waits for that request.

ws: WebSocketProtocol | None = None

The web socket used by the client to communicate with the server.

routes: dict[str, collections.abc.Callable[..., Union[Any, collections.abc.Coroutine[Any, Any, Any]]]]
listeners: defaultdict[str, list[collections.abc.Callable[..., Union[Any, collections.abc.Coroutine[Any, Any, Any]]]]]
connections: SimpleAttrDict[ConnectionT]
property loop: AbstractEventLoop

The event loop used by the client.

add_listener(func: Callable[[...], Any | Coroutine[Any, Any, Any]], name: str | None = None) → None

Add a function (called a listener) that executes the event when it occurs. The default available events can be found in the EventReference menu.

Parameters:

  • func – Function to be executed when an event occurs.

  • name – The name of what event to listen for. If None, the function name is used instead.

remove_listener(target: Callable[[...], Any | Coroutine[Any, Any, Any]] | str) → None

The name of what event to listen for. If None, the function name is used instead.

Parameters:

target – The name or function of the listener to be deleted.

listen(name: str | None = None) → Callable[[LiT], LiT]

Decorator version of add_listener().

Parameters:

name – The name of what event to listen for. If None, the function name is used instead.

dispatch(name: str, *args: Any, **kwargs: Any) → None

Raises an event.

Parameters:

  • name – The name of the event.

  • *args – The arguments to be passed to the event handler (listener).

  • **kwargs – The keyword arguments to be passed to the event handler (listener).

set_route(func: Callable[[...], Any | Coroutine[Any, Any, Any]], name: str | None = None, secret: bool = False) → None

Registers a function to respond to a request from another client. Now the registered function can be called by another client with the name used for registration. (This is called a request.) Such a function is also called a str.

Parameters:

  • func – Function to register as str.

  • name – str Name. If None, the function name is used.

  • secret – If it is True, function is to be registered as a hidden str. This is used internally in ipcs to add a str to receive information when a client connects from the server to a new connection, for example. A str registered in this way cannot be deleted. Therefore, it should not be used outside of ipcs.

remove_route(target: Callable[[...], Any | Coroutine[Any, Any, Any]] | str) → None

Delete a str.

Parameters:

target – Name or function of str to be deleted.

route(name: str | None = None, secret: bool = False) → Callable[[RhReT], RhReT]

Decorator version of set_route().

Parameters:
generate_session() → str

Generates a session ID used to identify the request. This is used inside ipcs.

async request(id_: str, route: str, *args: Any, **kwargs: Any) → Any

Sends a request to the specified connection. Alias of ipcs.connection.Connection.request().

Parameters:

  • id – The id of the connection.

  • route – The route to request.

  • *args – The arguments to be passed to str.

  • **kwargs – The keyword arguments to be passed to str.

Raises:
async request_all(route: str, *args: ~typing.Any, key_: ~collections.abc.Callable[[~ipcs.connection.Connection], bool] = <function AbcClient.<lambda>>, **kwargs: ~typing.Any) → dict[str, tuple[Optional[Any], None | ipcs.errors.RequestError]]

Make requests to everyone except yourself.

Parameters:

  • route – The name of the str to request.

  • *args – The arguments to be passed to str.

  • key – Function executed to determine if a request should be sent. This can be used, for example, to send a request only to IDs that conform to certain rules.

  • **kwargs – The keyword arguments to be passed to str.

Returns:

The return value is a dictionary whose key is the ID of the requestor and whose value is a tuple containing the return value and the error. If no error occurred, the error value is None.

abstract async start(*args: Any, **kwargs: Any) → None

This method is not fully implemented. Implemented are ipcs.client.Client and ipcs.server.Server.

abstract async close(code: int = 1000, reason: str = '...') → None

This method is not fully implemented. Implemented are ipcs.client.Client and ipcs.server.Server.

run(*args: Any, **kwargs: Any) → None

Start client. It internally uses run in the standard library asyncio to run ipcs.AbcClient.start().

Parameters:
class ipcs.Client(*args: Any, **kwargs: Any)

Bases: AbcClient[Connection]

This is the class of the client for ipc communication with the server.

ready: Event

This is an Event in the standard library asyncio to put if it has already started (connected to the server). You can use await ready.wait() to wait until it starts.

routes: dict[str, collections.abc.Callable[..., Union[Any, collections.abc.Coroutine[Any, Any, Any]]]]
listeners: defaultdict[str, list[collections.abc.Callable[..., Union[Any, collections.abc.Coroutine[Any, Any, Any]]]]]
connections: SimpleAttrDict[ConnectionT]
async start(*args: Any, reconnect: bool = True, **kwargs: Any) → None

Starts (connects) client.

Parameters:

  • *args –

    The arguments to be passed to connect in the third party library websockets. Details of connect is here.

  • reconnect – Whether or not to make the reconnection.

  • **kwargs –

    The keyword arguments to be passed to connect in the third party library websockets. Details of connect is here.

async close(code: int = 1000, reason: str = '...') → None

Disconnects the client from the server.

Parameters:

  • code – This code is used when cutting with a websocket.

  • reason – This reason is used when cutting with a websocket.

class ipcs.Connection(client: AbcClient, id_: str)

Bases: object

This class is used to represent the connection to the client. Requests to the client can be made using this.

Parameters:

  • client – This is the client who established this connection.

  • id – The ID of the client connected to this connection.

client

This is the client who established this connection.

id_

The ID of the client connected to this connection.

async close() → None

Disconnect this connection.

async request(route: str, *args: Any, ipcs_secret: bool = False, **kwargs: Any) → Any

Make a request to the client connected by this connection.

Parameters:

  • route – The name of the str you wish to execute at the request destination.

  • *args – The arguments to be passed to the str.

  • ipcs_secret – Whether the request is for internal use by ipcs. Do not set this to True for normal use.

  • **kwargs – The arguments to be passed to the str.

Raises:
class ipcs.Server(*args: Any, **kwargs: Any)

Bases: AbcClient[ConnectionForServer | Connection]

Server for ipcs communication. This is not normally used to run ipcs servers. Instead, use ipcs-server, which is available on the command line. This command is automatically set up during installation with pip.

async communicate(ws: WebSocketProtocol) → None

Function called when a new client connects. Used internally. ipcs uses web sockets using a third-party library called websocekts, but if you want to use another web socket library, pass the newly connected web socket to this function. However, since the websockets available to ipcs must conform to ipcs.types_.WebSocketProtocol, you will need to wrap the websocket class to make it conform to ipcs.types_.WebSocketProtocol if it differs from that.

Parameters:

ws – The websocket of the newly connected client.

async start(*args: Any, **kwargs: Any) → None

Run the server.

Parameters:

  • *args –

    The arguments to be passed to serve in the third party library websockets. Details of serve in here.

  • **kwargs –

    The keyword arguments to be passed to serve in the third party library websockets. Details of serve in here.

async close(code: int = 1000, reason: str = '...') → None

Terminate the server.

Parameters:

  • code – This code is used when cutting with a websocket.

  • reason – This reason is used when cutting with a websocket.

class ipcs.ConnectionForServer(client: AbcClient, id_: str)

Bases: Connection

Class that extends Connection for Server.

Parameters:

ws – This is the websocket to the client connected by this connection.

ws: WebSocketProtocol

This is the websocket to the client connected by this connection.

async close() → None

Disconnect this connection.

queue: dict[str, DataEvent[ResponsePayload]]
exception ipcs.IpcsError

Bases: Exception

All errors defined in ipcs are inherited errors.

exception ipcs.IdIsNotFound

Bases: IpcsError

This error occurs when str is not found.

exception ipcs.RequestError

Bases: IpcsError

This error occurs when a request fails.

exception ipcs.FailedToProcessError(message: str, error: str, *args: Any, **kwargs: Any)

Bases: RequestError

Occurs when an error occurs at the request destination.

error

The name and description of the error that occurred at the request destination. The format is <name>: <content>.

exception ipcs.ClosedConnectionError

Bases: RequestError

This occurs when a request is attempted but communication is lost.