Skip to content

Base

pyvider.rpcplugin.transport.base

Base Abstract Class for RPC Plugin Transports.

This module defines RPCPluginTransport, an abstract base class (ABC) that outlines the contract for all transport implementations within the Pyvider RPC Plugin system. Concrete transport classes (e.g., for TCP or Unix sockets) must inherit from this class and implement its abstract methods.

Classes

RPCPluginTransport

Bases: ABC

Abstract base class defining the interface for all transport implementations.

This class defines the contract that concrete transport implementations must fulfill to provide network communication for plugins. The interface supports both client-side (connect) and server-side (listen) operations.

Implementations must handle: - Connection setup and teardown - Socket lifecycle management - Error handling and reporting - Resource cleanup

Custom transports can be implemented by subclassing this class and implementing the required abstract methods.

Functions
close abstractmethod async 
close() -> None

Close the transport and release any associated resources.

Implementations should ensure that all network resources (like sockets) are properly closed and cleaned up. This method should be idempotent.

Raises:

Type Description
TransportError

If an error occurs during closing.
Source code in pyvider/rpcplugin/transport/base.py 
72
73
74
75
76
77
78
79
80
81
82
83
@abc.abstractmethod
async def close(self) -> None:  # pragma: no cover
    """
    Close the transport and release any associated resources.

    Implementations should ensure that all network resources (like sockets)
    are properly closed and cleaned up. This method should be idempotent.

    Raises:
        TransportError: If an error occurs during closing.
    """
    ...
connect abstractmethod async 
connect(endpoint: str) -> None

Connect to a remote endpoint.

Implementations should establish a connection to the specified endpoint address. This is typically used by client components.

Parameters:

Name Type Description Default
endpoint str

The target endpoint address string.

 required

Raises:

Type Description
TransportError

If the connection cannot be established.
Source code in pyvider/rpcplugin/transport/base.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@abc.abstractmethod
async def connect(self, endpoint: str) -> None:  # pragma: no cover
    """
    Connect to a remote endpoint.

    Implementations should establish a connection to the specified endpoint
    address. This is typically used by client components.

    Args:
        endpoint: The target endpoint address string.

    Raises:
        TransportError: If the connection cannot be established.
    """
    ...
listen abstractmethod async 
listen() -> str

Start listening for connections.

Implementations should bind to an appropriate socket or address and begin accepting connections. This is typically used by server components.

Returns:

Type Description
str

The endpoint address as a string (e.g., "127.0.0.1:50051" or
str

"/tmp/socket.sock")

Raises:

Type Description
TransportError

If binding or listening fails
Source code in pyvider/rpcplugin/transport/base.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@abc.abstractmethod
async def listen(self) -> str:  # pragma: no cover
    """
    Start listening for connections.

    Implementations should bind to an appropriate socket or address and
    begin accepting connections. This is typically used by server components.

    Returns:
        The endpoint address as a string (e.g., "127.0.0.1:50051" or
        "/tmp/socket.sock")

    Raises:
        TransportError: If binding or listening fails
    """
    ...