Client Pairing¶

Module: s2auth.client.pairing

`LOGGER = logging.getLogger(name)` `module-attribute` ¶

`HttpRequestHook = Callable[[httpx.Request], Awaitable[None]]` `module-attribute` ¶

`HttpResponseHook = Callable[[httpx.Response], Awaitable[None]]` `module-attribute` ¶

`HttpEventHooks = pairing_core.HttpEventHooks` `module-attribute` ¶

`ConnectionStore` ¶

Bases: Protocol

Storage interface used by pairing client flows.

Source code in src/s2auth/client/connection_store.py

class ConnectionStore(Protocol):
    """Storage interface used by pairing client flows."""

    def store_connection_details(self, s2_node_id: str, details: dict[str, Any]) -> None:
        ...

    def load_connection_details(self, s2_node_id: str) -> dict[str, Any] | None:
        ...

    def remove_connection_details(self, s2_node_id: str) -> bool:
        ...

`Dao` ¶

SQLAlchemy-backed data access object for storing/loading connection details. Default database is SQLite file 'connection_details.db'.

Source code in src/s2auth/client/dao.py

class Dao:
    """
    SQLAlchemy-backed data access object for storing/loading connection details.
    Default database is SQLite file 'connection_details.db'.
    """

    def __init__(self, db_url: str = "sqlite:///connection_details.db") -> None:
        # Create engine & session factory
        self._engine = create_engine(db_url, future=True)
        Base.metadata.create_all(self._engine)

        self._SessionLocal = sessionmaker(
            bind=self._engine,
            autoflush=False,
            autocommit=False,
            future=True,
        )

    def store_connection_details(self, s2_node_id: str, details: dict[str, Any]) -> None:
        """
        Insert or overwrite a connection detail identified by s2_node_id.
        """
        with self._SessionLocal() as session:
            with session.begin():
                obj = session.query(ConnectionDetail).filter(ConnectionDetail.s2_node_id == s2_node_id).one_or_none()

                if obj is None:
                    obj = ConnectionDetail(s2_node_id=s2_node_id)
                    session.add(obj)

                for detail_key, model_attr in details.items():
                    if hasattr(ConnectionDetail, detail_key):
                        setattr(obj, detail_key, model_attr)
                    else:
                        raise ValueError(f"Invalid detail key: {detail_key}")

    def load_connection_details(self, s2_node_id: str) -> Optional[dict[str, Any]]:
        """Load the full connection details object for the given node ID."""
        stmt: Select[Any] = (
            select(ConnectionDetail)
            .where(ConnectionDetail.s2_node_id == s2_node_id)
            .limit(1)
        )
        with self._SessionLocal() as session:
            obj = session.execute(stmt).scalars().first()
            if obj is None:
                return None
            return {c.key: getattr(obj, c.key) for c in inspect(obj).mapper.column_attrs}

    def remove_connection_details(self, s2_node_id: str) -> bool:
        """Remove connection details for the given node ID.

        Returns True when an entry was deleted, otherwise False.
        """
        stmt: Select[Any] = (
            select(ConnectionDetail)
            .where(ConnectionDetail.s2_node_id == s2_node_id)
            .limit(1)
        )
        with self._SessionLocal() as session:
            with session.begin():
                obj = session.execute(stmt).scalars().first()
                if obj is None:
                    return False
                session.delete(obj)
            return True

`store_connection_details(s2_node_id, details)` ¶

Insert or overwrite a connection detail identified by s2_node_id.

Source code in src/s2auth/client/dao.py

def store_connection_details(self, s2_node_id: str, details: dict[str, Any]) -> None:
    """
    Insert or overwrite a connection detail identified by s2_node_id.
    """
    with self._SessionLocal() as session:
        with session.begin():
            obj = session.query(ConnectionDetail).filter(ConnectionDetail.s2_node_id == s2_node_id).one_or_none()

            if obj is None:
                obj = ConnectionDetail(s2_node_id=s2_node_id)
                session.add(obj)

            for detail_key, model_attr in details.items():
                if hasattr(ConnectionDetail, detail_key):
                    setattr(obj, detail_key, model_attr)
                else:
                    raise ValueError(f"Invalid detail key: {detail_key}")

`load_connection_details(s2_node_id)` ¶

Load the full connection details object for the given node ID.

Source code in src/s2auth/client/dao.py

def load_connection_details(self, s2_node_id: str) -> Optional[dict[str, Any]]:
    """Load the full connection details object for the given node ID."""
    stmt: Select[Any] = (
        select(ConnectionDetail)
        .where(ConnectionDetail.s2_node_id == s2_node_id)
        .limit(1)
    )
    with self._SessionLocal() as session:
        obj = session.execute(stmt).scalars().first()
        if obj is None:
            return None
        return {c.key: getattr(obj, c.key) for c in inspect(obj).mapper.column_attrs}

`remove_connection_details(s2_node_id)` ¶

Remove connection details for the given node ID.

Returns True when an entry was deleted, otherwise False.

Source code in src/s2auth/client/dao.py

def remove_connection_details(self, s2_node_id: str) -> bool:
    """Remove connection details for the given node ID.

    Returns True when an entry was deleted, otherwise False.
    """
    stmt: Select[Any] = (
        select(ConnectionDetail)
        .where(ConnectionDetail.s2_node_id == s2_node_id)
        .limit(1)
    )
    with self._SessionLocal() as session:
        with session.begin():
            obj = session.execute(stmt).scalars().first()
            if obj is None:
                return False
            session.delete(obj)
        return True

`ClientSettings` ¶

Bases: BaseSettings

Settings for the high-level S2 pairing client API.

Values can be provided directly or loaded from environment variables/.env files.

Source code in src/s2auth/client/settings.py

class ClientSettings(BaseSettings):
    """Settings for the high-level S2 pairing client API.

    Values can be provided directly or loaded from environment variables/.env files.
    """

    model_config = SettingsConfigDict(
        env_file=[".env", ".env.docker"],
        extra="ignore",
        env_nested_delimiter="__",
    )

    server_url: str = "http://localhost"
    pairing_token: str | None = None
    pairing_s2_node_id: str | None = None
    client_s2_node_id: str | None = None

    client_role: Role = Role.RM
    client_deployment: Deployment | None = None
    domain_name: str | None = None

    verify_tls: bool = True
    ssl_certfile: str | None = Field(default=None)
    storage_db_url: str = "sqlite:///connection_details.db"

    supported_s2_versions: list[str] = Field(default_factory=lambda: ["v1"])
    supported_communication_protocols: list[CommunicationProtocol] = Field(
        default_factory=lambda: [CommunicationProtocol.WebSocket]
    )
    supported_hmac_hashing_algorithms: list[HmacHashingAlgorithm] = Field(
        default_factory=lambda: [HmacHashingAlgorithm.SHA256]
    )

    cleint_brand: str = "ExampleHeatCo"
    client_device_type: str = "Heatpump"
    client_model_name: str = "SmartHeatPump X200"

`CommunicationProtocol` ¶

Bases: str, Enum

Source code in src/s2auth/common/model/s2_connect_common.py

class CommunicationProtocol(str, Enum):
    WebSocket = "WebSocket"

`Deployment` ¶

Bases: str, Enum

Source code in src/s2auth/common/model/s2_connect_common.py

class Deployment(str, Enum):
    WAN = "WAN"
    LAN = "LAN"

`NodeDescription` ¶

Bases: BaseModel

Source code in src/s2auth/common/model/s2_connect_common.py

class NodeDescription(BaseModel):
    id: NodeId
    brand: str
    logoUrl: AnyUrl | None = None
    type: str
    modelName: str
    userDefinedName: str | None = None
    role: Role

`NodeId` ¶

Bases: RootModel[UUID]

Source code in src/s2auth/common/model/s2_connect_common.py

class NodeId(RootModel[UUID]):
    root: Annotated[UUID, Field(description="Unique identifier of the node")]

`Role` ¶

Bases: str, Enum

Source code in src/s2auth/common/model/s2_connect_common.py

class Role(str, Enum):
    CEM = "CEM"
    RM = "RM"

`HmacHashingAlgorithm` ¶

Bases: str, Enum

Source code in src/s2auth/common/model/s2_connect_pairing.py

class HmacHashingAlgorithm(str, Enum):
    SHA256 = "SHA256"