mercury-ocip-fast

@attrs.define(kw_only=True)
class Client[S: (TCPSessionSettings, SOAPSessionSettings)]:
    """An OCI-P Client for a single User.

    The client opens sessions, it logs them in as the specified user,
    and sends commands over them.

    The client needs an async setup step before use. Make the client
    with the ``create`` method, or use it in an ``async with`` block.

    Attributes:
        host: The host name or address of the BroadWorks server.
        port: The port of the server. If None, the endpoint or the
            scheme sets the port.
        username: The user name for the login.
        password: The password for the login. Treat this value as a
            secret.
        atom_type: The class of session to open, for example the SOAP
            atom or the TCP atom.
        session_config: The transport settings for each session, for
            example the timeouts.
        pool_config: The settings for the session pool, for example the
            maximum size.
        user_agent: The user agent name for the client.
        tls: If true, use a TLS link. A TLS link protects the password,
            so the client uses the plain-text login. If false, the
            client uses the encrypted login.
    """

    host: str
    port: int | None = None
    username: str
    password: str
    atom_type: type[SessionAtom[S]]
    session_config: S = attrs.field()
    pool_config: SessionPoolSettings = attrs.field()
    tls: bool = True
    logger: logging.Logger = attrs.field(default=logging.getLogger(__name__))
    _requester: Requester = attrs.field(init=False)
    _pool: SessionPool[SessionAtom[S]] = attrs.field(init=False)
    _authenticator: Authenticator = attrs.field(init=False)

    async def _async_setup(self) -> Self:
        """Setup the client and its dependencies.

        Returns:
            The client
        """
        if getattr(self, "_pool", None) is not None:
            return self

        self.logger.info(
            f"Initializing requester for {self.host}:{self.port} (tls={self.tls})"
        )
        self._requester = Requester()
        self._authenticator = Authenticator(
            username=self.username, password=self.password, requester=self._requester
        )
        self._pool = SessionPool(
            default_factory=self._default_factory,
            pool_settings=self.pool_config,
        )

        return self

    async def _default_factory(self) -> SessionAtom[S]:
        """Create a SessionAtom and log it in based on the client's TLS value."""
        atom = await self.atom_type.open(
            self.host,
            self.port,
            settings=self.session_config,
            verify_ssl=self.tls,
        )

        try:
            if (
                self.tls
            ):  # TLS protects the password, so a TLS link uses the plain-text login.
                await self._authenticator.generic_login(atom)
            else:
                await self._authenticator.encrypted_login(atom)
        except BaseException:
            await atom.close()
            raise

        return atom

    async def __aenter__(self) -> Self:
        """Do the async setup at the start of an ``async with`` block.

        Returns:
            The client, now ready for use.
        """
        return await self._async_setup()

    async def __aexit__(
        self, _exc_type: object, _exc_val: object, _exc_tb: object
    ) -> None:
        """Close the client at the end of an ``async with`` block.

        This method closes the client for each exit. It closes the
        client after an error, and it closes the client after a normal
        exit.
        """
        await self.close()

    @classmethod
    async def create(cls, **kwargs) -> Self:
        """Make a client and do the async setup.

        Use this method to get a client without an ``async with`` block.
        Keep the client, use it, and then close it with the ``close``
        method.

        Args:
            kwargs: The keyword arguments for the client, for example
                the host, the user name, and the password.

        Returns:
            A client that is ready for use.
        """
        self = cls(**kwargs)
        return await self._async_setup()

    @overload
    async def command[R: OCIResponse](self, request: OCIRequest[R]) -> R: ...

    @overload
    async def command[R: OCIResponse](
        self, request: list[OCIRequest[R]]
    ) -> list[R]: ...

    @overload
    async def command[R: OCIResponse](
        self, request: OCIRequest, *, response_type: type[R]
    ) -> R: ...

    @overload
    async def command[R: OCIResponse](
        self, request: list[OCIRequest], *, response_type: type[R] | list[R]
    ) -> list[R]: ...

    async def command[R: OCIResponse](
        self,
        request: OCIRequest[R] | list[OCIRequest[R]],
        *,
        response_type: type[R] | None = None,
    ) -> R | list[R]:
        """Send one command, or a batch, and return the parsed response(s).

        The result is typed as the request's own response class, resolved
        from ``OCIRequest[R]``. Pass ``response_type`` only to override that
        with an explicit class.

        Args:
            request: One OCI request, or a list of requests for a batch.
            response_type: The class to parse each response into. If None,
                each request's ``_response_cls`` is used.

        Returns:
            The parsed response, or a list of responses for a batch.
        """
        all_results: list[R] = []

        async with self._pool.session() as atom:
            if isinstance(request, OCIRequest):
                return await self._requester.send(
                    payload=request.to_xml(),
                    response_type=response_type or request._response_cls,
                    session=atom,
                )

            for batch in batched(request, 15):
                result = await self._requester.send(
                    payload=[b.to_xml() for b in batch],
                    response_type=response_type or [b._response_cls for b in batch],
                    session=atom,
                )
                all_results.extend(result)

            return all_results

    async def close(self) -> None:
        """Close the client and every session in the pool.

        This method closes the session pool. The pool closes each
        session and lets go of each transport. The method is safe to
        call more than once.
        """
        pool = getattr(self, "_pool", None)  # Survives half constructed client
        if pool is not None:
            await pool.close()

@attrs.define(kw_only=True)
class SessionClient:
    """OCI-P client.

    A multi-tenant client. It holds no identity and keeps no session list.
    The client shares one requester across all sessions; each opened session is
    owned by its caller and may be passed around, closed, or exported and later
    resumed.

    Set up the client with ``create()`` or an ``async with`` block.

    Attributes:
        host: BroadWorks server hostname or address.
        port: BroadWorks server port. If ``None``, the endpoint or scheme
            supplies it.
        atom_type: Resumable session class to open, such as the SOAP atom.
        session_config: Transport settings for each session.
        tls: Whether to use TLS. If true, the client uses plain-text login; if
            false, it uses the encrypted login.
    """

    host: str
    port: int | None = None
    atom_type: type[SoapAtom]
    session_config: SOAPSessionSettings = attrs.field()
    tls: bool = True
    logger: logging.Logger = attrs.field(default=logging.getLogger(__name__))
    _requester: Requester = attrs.field(init=False)

    async def _async_setup(self) -> Self:
        """Set up the shared requester. Idempotent.

        Returns:
            The client.
        """
        if getattr(self, "_requester", None) is not None:
            return self

        self.logger.info(
            f"Initializing session client for {self.host}:{self.port} (tls={self.tls})"
        )
        self._requester = Requester()

        return self

    async def _login_factory(self, username: str, password: str) -> SoapAtom:
        """Open a session and log it in as the given user.

        Each caller brings its own credentials, so the authenticator is
        built per login rather than held on the client.
        """
        atom = await self.atom_type.open(
            self.host,
            self.port,
            settings=self.session_config,
            verify_ssl=self.tls,
        )

        authenticator = Authenticator(
            username=username, password=password, requester=self._requester
        )
        try:
            if (
                self.tls
            ):  # TLS protects the password, so a TLS link uses the plain-text login.
                await authenticator.generic_login(atom)
            else:
                await authenticator.encrypted_login(atom)
        except BaseException:
            await atom.close()
            raise

        return atom

    async def _resume_factory(self, pair: SessionPair) -> SoapAtom:
        """Open a session from a stored pair, without a fresh login."""
        return await self.atom_type.resume(
            self.host,
            pair,
            settings=self.session_config,
            verify_ssl=self.tls,
        )

    async def __aenter__(self) -> Self:
        """Do the async setup at the start of an ``async with`` block.

        Returns:
            The client, now ready for use.
        """
        return await self._async_setup()

    async def __aexit__(
        self, _exc_type: object, _exc_val: object, _exc_tb: object
    ) -> None:
        """Leave the ``async with`` block.

        The client owns no sessions, so it has nothing to close here. The
        caller must close each session it opened.
        """

    @classmethod
    async def create(cls, **kwargs) -> Self:
        """Make a client and do the async setup, without an ``async with``.

        Args:
            kwargs: The keyword arguments for the client, for example the
                host and the atom type.

        Returns:
            A client that is ready for use.
        """
        self = cls(**kwargs)
        return await self._async_setup()

    async def open(self, username: str, password: str) -> SoapAtom:
        """Open a fresh session, logged in as ``username``.

        The returned atom is the caller's handle. Pass it to ``command`` and
        close it with ``close`` when done.

        Args:
            username: The user name for the login.
            password: The password for the login. Treat this as a secret.

        Returns:
            A logged-in session, owned by the caller.

        Raises:
            MErrorLogin: If the server rejects the login.
        """
        return await self._login_factory(username, password)

    async def resume(self, pair: SessionPair) -> SoapAtom:
        """Open a session from a stored pair, resuming an earlier login.

        Use the atom's ``pair`` property to export the identity for later.

        Args:
            pair: The stored identity from an earlier session.

        Returns:
            A resumed session, owned by the caller.
        """
        return await self._resume_factory(pair)

    @overload
    async def command[R: OCIResponse](
        self, session: SoapAtom, request: OCIRequest[R]
    ) -> R: ...

    @overload
    async def command[R: OCIResponse](
        self, session: SoapAtom, request: list[OCIRequest[R]]
    ) -> list[R]: ...

    @overload
    async def command[R: OCIResponse](
        self, session: SoapAtom, request: OCIRequest, *, response_type: type[R]
    ) -> R: ...

    @overload
    async def command[R: OCIResponse](
        self, session: SoapAtom, request: list[OCIRequest], *, response_type: type[R]
    ) -> list[R]: ...

    async def command[R: OCIResponse](
        self,
        session: SoapAtom,
        request: OCIRequest[R] | list[OCIRequest[R]],
        *,
        response_type: type[R] | None = None,
    ) -> R | list[R]:
        """Send one command, or a batch, over the caller's session.

        The result is typed as the request's own response class, resolved
        from ``OCIRequest[R]``. Pass ``response_type`` only to override that
        with an explicit class.

        Args:
            session: The caller's session, from ``open`` or ``resume``.
            request: One OCI request, or a list of requests for a batch.
            response_type: The class to parse each response into. If None,
                each request's ``_response_cls`` is used.

        Returns:
            The parsed response, or a list of responses for a batch.
        """
        all_results: list[R] = []

        if isinstance(request, OCIRequest):
            return await self._requester.send(
                payload=request.to_xml(),
                response_type=response_type or request._response_cls,
                session=session,
            )

        for batch in batched(request, 15):
            result = await self._requester.send(
                payload=[b.to_xml() for b in batch],
                response_type=response_type or [b._response_cls for b in batch],
                session=session,
            )
            all_results.extend(result)

        return all_results

    async def close(self, session: SoapAtom) -> None:
        """Close one of the caller's sessions and let go of its transport.

        Args:
            session: The session to close.
        """
        await session.close()

@attrs.define(kw_only=True, slots=True)
class SOAPSessionAtom:
    """A SOAP session for BroadWorks.

    The transport is an httpx client. The client has its own cookie jar.
    The cookie jar holds the JSESSIONID cookie after a login.

    The session also has an OCI-P ``session_id``. This id goes in the body
    of each message. The class makes a new UUID for the id by default.

    Attributes:
        endpoint: The URL of the SOAP service.
        http_client: The httpx client. It holds this session's cookie jar.
        settings: The timeouts for the httpx client.
        session_id: The OCI-P session id for each message body.
    """

    endpoint: str
    http_client: httpx.AsyncClient
    settings: SOAPSessionSettings = attrs.field(default=SOAPSessionSettings())
    session_id: str = attrs.field(factory=lambda: str(uuid.uuid4()))
    created_at: float = attrs.field(factory=time.monotonic)
    last_used: float = attrs.field(factory=time.monotonic)

    @classmethod
    async def open(
        cls,
        endpoint: str,
        port: int | None = None,
        *,
        settings: SOAPSessionSettings,
        verify_ssl: bool = True,
    ) -> SOAPSessionAtom:
        """Make a new SOAP session. The session is not logged in.

        The new session has its own httpx client. The client has no
        cookies. A login must run before the session can send commands.

        Args:
            endpoint: The full URL of the SOAP service. The URL can hold the
                port, for example ``https://host:8443/webservice``.
            port: The port of the service. This argument has priority over a
                port in the URL. If None, the URL, or the scheme, sets the
                port.
            settings: The timeouts for the httpx client.
            verify_ssl: If true, verify the TLS certificate of the server.

        Returns:
            A new SOAP session.

        Raises:
            MErrorHttpInitialisation: If the URL has no host.
        """
        try:
            url = override_url_port(endpoint, port)
        except ValueError as e:
            raise MErrorHttpInitialisation(str(e)) from e

        timeout = httpx.Timeout(
            connect=settings.connect_timeout,
            read=settings.read_timeout,
            write=settings.write_timeout,
            pool=None,
        )

        logger.debug(
            "Open a SOAP session to %s, connect timeout %ss, read timeout %ss",
            url,
            settings.connect_timeout,
            settings.read_timeout,
        )

        return cls(
            endpoint=url,
            http_client=httpx.AsyncClient(
                verify=verify_ssl,
                timeout=timeout,
            ),
            settings=settings,
        )

    @classmethod
    async def resume(
        cls,
        endpoint: str,
        pair: SessionPair,
        *,
        settings: SOAPSessionSettings,
        verify_ssl: bool = True,
    ) -> SOAPSessionAtom:
        """Make a SOAP session from a stored session pair.

        Use this method to continue a session after a restart. The new
        session takes the JSESSIONID cookie and the OCI-P session id from
        the pair. The server then accepts the session as logged in.

        Args:
            endpoint: The URL of the SOAP service.
            pair: The stored identity. It holds the cookie and the id.
            settings: The timeouts for the httpx client.
            verify_ssl: If true, verify the TLS certificate of the server.

        Returns:
            A SOAP session that uses the given pair.
        """
        session = await cls.open(endpoint, settings=settings, verify_ssl=verify_ssl)
        session.http_client.cookies.set("JSESSIONID", pair.jsessionid)
        session.session_id = pair.session_id

        logger.debug("Resume a SOAP session to %s from a stored pair", endpoint)

        return session

    @property
    def jsessionid(self) -> str | None:
        """The JSESSIONID cookie of this session.

        The value is None before the login.

        Do not write this value to a log. It is a secret. A person who has
        this value can send commands as the user.
        """
        return self.http_client.cookies.get("JSESSIONID")

    @property
    def pair(self) -> SessionPair:
        """The session pair of this session. Store it to resume later.

        Raises:
            MErrorMissingSessionIdentity: If the session has no login yet.
        """
        jsessionid = self.jsessionid
        if not jsessionid:
            raise MErrorMissingSessionIdentity()
        return SessionPair(jsessionid=jsessionid, session_id=self.session_id)

    async def send(self, payload: str | list[str]) -> str:
        """Send one envelope to the server and return one reply.

        This method wraps the payload in the BroadsoftDocument and then in
        the SOAP envelope. It posts the envelope with the httpx client. The
        JSESSIONID cookie goes with the request. The method then gets the
        OCI reply from the response.

        Do not write the payload or the reply to a log. They can hold
        secrets, for example a password in a login command.

        Args:
            payload: One OCI command, or a list of OCI commands.

        Returns:
            The OCI reply as a string.

        Raises:
            MErrorHttpStatus: If the server returns a non-2xx HTTP status.
            MErrorHttpTimeout: If the request does not complete in time.
            MErrorHttpInitialisation: If the client cannot connect.
            MErrorHttpDropped: If the connection stops during the request.
        """
        oci_xml = build_broadsoft_envelope(payload, self.session_id)
        soap_envelope = wrap_soap(oci_xml)

        logger.debug("Send %d bytes to %s", len(soap_envelope), self.endpoint)

        try:
            response = await self.http_client.post(
                self.endpoint,
                content=soap_envelope.encode("utf-8"),
                headers={"Content-Type": "text/xml; charset=UTF-8", "SOAPAction": ""},
            )
            response.raise_for_status()
        except httpx.HTTPStatusError as e:
            logger.warning(
                "The server %s returned HTTP %d", self.endpoint, e.response.status_code
            )
            raise MErrorHttpStatus(
                f"BroadWorks returned HTTP {e.response.status_code}",
                status=e.response.status_code,
            ) from e
        except httpx.TimeoutException as e:
            logger.warning("The request to %s timed out: %s", self.endpoint, e)
            raise MErrorHttpTimeout(str(e)) from e
        except httpx.ConnectError as e:
            logger.warning("Cannot connect to %s: %s", self.endpoint, e)
            raise MErrorHttpInitialisation(str(e)) from e
        except httpx.TransportError as e:  # a read, write, or protocol error
            logger.warning(
                "The connection to %s dropped mid-request: %s", self.endpoint, e
            )
            raise MErrorHttpDropped(str(e)) from e

        logger.debug(
            "Receive %d bytes from %s, HTTP %d",
            len(response.text),
            self.endpoint,
            response.status_code,
        )

        return unwrap_soap(response.text)

    async def close(self) -> None:
        """Close the httpx client.

        This closes the cookie jar and the open sockets. The method is safe
        to call more than once.
        """
        logger.debug("Close the SOAP session to %s", self.endpoint)
        try:
            await self.http_client.aclose()
        except Exception as e:  # noqa: BLE001
            logger.warning(
                "The connection to %s raised an exception while closing: %s",
                self.endpoint,
                e,
            )

    def is_alive(self) -> bool:
        """Tell if the session is still connected."""
        return not self.http_client.is_closed

    def is_stale(self) -> bool:
        """Tell if the session is past its time to live."""
        return (time.monotonic() - self.created_at) > self.settings.max_ttl_seconds