client.py

# This file was auto-generated by Fern from our API Definition.

from __future__ import annotations

import datetime as dt
import typing

from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper
from ..core.request_options import RequestOptions
from ..types.grants_archive_response import GrantsArchiveResponse
from ..types.grants_create_response import GrantsCreateResponse
from ..types.grants_exchange_response import GrantsExchangeResponse
from ..types.grants_get_response import GrantsGetResponse
from ..types.grants_list_response import GrantsListResponse
from ..types.grants_update_response import GrantsUpdateResponse
from ..types.identity import Identity
from .raw_client import AsyncRawGrantsClient, RawGrantsClient
from .types.grants_create_request_layer_mode import GrantsCreateRequestLayerMode
from .types.grants_create_request_visitor_identity import GrantsCreateRequestVisitorIdentity
from .types.grants_exchange_request_visitor_identity import GrantsExchangeRequestVisitorIdentity
from .types.grants_update_request_layer_mode import GrantsUpdateRequestLayerMode
from .types.grants_update_request_visitor_identity import GrantsUpdateRequestVisitorIdentity

if typing.TYPE_CHECKING:
    from .visitors.client import AsyncVisitorsClient, VisitorsClient
# this is used as the default value for optional parameters
OMIT = typing.cast(typing.Any, ...)


class GrantsClient:
    def __init__(self, *, client_wrapper: SyncClientWrapper):
        self._raw_client = RawGrantsClient(client_wrapper=client_wrapper)
        self._client_wrapper = client_wrapper
        self._visitors: typing.Optional[VisitorsClient] = None

    @property
    def with_raw_response(self) -> RawGrantsClient:
        """
        Retrieves a raw implementation of this client that returns raw responses.

        Returns
        -------
        RawGrantsClient
        """
        return self._raw_client

    def exchange(
        self,
        *,
        grant: str,
        passphrase: typing.Optional[str] = OMIT,
        visitor_handle: typing.Optional[str] = OMIT,
        restart: typing.Optional[bool] = OMIT,
        visitor_identity: typing.Optional[GrantsExchangeRequestVisitorIdentity] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> GrantsExchangeResponse:
        """
        Unauthenticated, but requires a browser Origin header, checked against the grant allowlist. Unknown, deleted, paused and archived grants are indistinguishable (404). Passphrase-protected grants return 422 GrantPassphraseRequired until `passphrase` is supplied. A grant that asks its visitors who they are on open returns 422 VisitorIdentityRequired until the required answers are given; `error.details` carries `fields` (what the grant asks), `missing`, `invalid`, `purpose` and `privacyUrl`. A grant that asks before the first change issues the session at once with `visitor.identityRequest`, and writes answer 403 VisitorIdentityRequired with the same details until the visitor exchanges again with their answers. Answers that can not be taken answer 422 on any grant. Mounted only when the deployment can sign (HS256 mode).

        Parameters
        ----------
        grant : str

        passphrase : typing.Optional[str]

        visitor_handle : typing.Optional[str]

        restart : typing.Optional[bool]

        visitor_identity : typing.Optional[GrantsExchangeRequestVisitorIdentity]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GrantsExchangeResponse
            OK

        Examples
        --------
        from cloudpdf import CloudPDFClient

        client = CloudPDFClient(
            token="YOUR_TOKEN",
        )
        client.grants.exchange(
            grant="grant",
        )
        """
        _response = self._raw_client.exchange(
            grant=grant,
            passphrase=passphrase,
            visitor_handle=visitor_handle,
            restart=restart,
            visitor_identity=visitor_identity,
            request_options=request_options,
        )
        return _response.data

    def list(
        self,
        tenant_id: str,
        *,
        limit: typing.Optional[int] = None,
        cursor: typing.Optional[str] = None,
        doc_id: typing.Optional[str] = None,
        archived: typing.Optional[bool] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> GrantsListResponse:
        """
        Parameters
        ----------
        tenant_id : str

        limit : typing.Optional[int]

        cursor : typing.Optional[str]

        doc_id : typing.Optional[str]

        archived : typing.Optional[bool]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GrantsListResponse
            OK

        Examples
        --------
        from cloudpdf import CloudPDFClient

        client = CloudPDFClient(
            token="YOUR_TOKEN",
        )
        client.grants.list(
            tenant_id="tenantId",
        )
        """
        _response = self._raw_client.list(
            tenant_id, limit=limit, cursor=cursor, doc_id=doc_id, archived=archived, request_options=request_options
        )
        return _response.data

    def create(
        self,
        tenant_id: str,
        *,
        doc_id: str,
        scope: typing.Sequence[str],
        name: typing.Optional[str] = OMIT,
        layer_mode: typing.Optional[GrantsCreateRequestLayerMode] = OMIT,
        layer_name: typing.Optional[str] = OMIT,
        keep_visitor_layers: typing.Optional[bool] = OMIT,
        visitor_idle_seconds: typing.Optional[int] = OMIT,
        visitor_identity: typing.Optional[GrantsCreateRequestVisitorIdentity] = OMIT,
        identity: typing.Optional[Identity] = OMIT,
        origins: typing.Optional[typing.Sequence[str]] = OMIT,
        passphrase: typing.Optional[str] = OMIT,
        session_ttl_seconds: typing.Optional[int] = OMIT,
        expires_at: typing.Optional[dt.datetime] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> GrantsCreateResponse:
        """
        The returned id is what a link page or an embed opens. Its scope gives only document permissions the caller holds itself, as tokens.issue does. Mounted only when the deployment can sign (HS256 mode) — exchange mints session JWTs, so grants exist only where minting does.

        Parameters
        ----------
        tenant_id : str

        doc_id : str

        scope : typing.Sequence[str]

        name : typing.Optional[str]

        layer_mode : typing.Optional[GrantsCreateRequestLayerMode]

        layer_name : typing.Optional[str]

        keep_visitor_layers : typing.Optional[bool]

        visitor_idle_seconds : typing.Optional[int]

        visitor_identity : typing.Optional[GrantsCreateRequestVisitorIdentity]

        identity : typing.Optional[Identity]

        origins : typing.Optional[typing.Sequence[str]]

        passphrase : typing.Optional[str]

        session_ttl_seconds : typing.Optional[int]

        expires_at : typing.Optional[dt.datetime]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GrantsCreateResponse
            OK

        Examples
        --------
        from cloudpdf import CloudPDFClient

        client = CloudPDFClient(
            token="YOUR_TOKEN",
        )
        client.grants.create(
            tenant_id="tenantId",
            doc_id="docId",
            scope=["scope"],
        )
        """
        _response = self._raw_client.create(
            tenant_id,
            doc_id=doc_id,
            scope=scope,
            name=name,
            layer_mode=layer_mode,
            layer_name=layer_name,
            keep_visitor_layers=keep_visitor_layers,
            visitor_idle_seconds=visitor_idle_seconds,
            visitor_identity=visitor_identity,
            identity=identity,
            origins=origins,
            passphrase=passphrase,
            session_ttl_seconds=session_ttl_seconds,
            expires_at=expires_at,
            request_options=request_options,
        )
        return _response.data

    def get(
        self, tenant_id: str, grant_id: str, *, request_options: typing.Optional[RequestOptions] = None
    ) -> GrantsGetResponse:
        """
        Parameters
        ----------
        tenant_id : str

        grant_id : str

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GrantsGetResponse
            OK

        Examples
        --------
        from cloudpdf import CloudPDFClient

        client = CloudPDFClient(
            token="YOUR_TOKEN",
        )
        client.grants.get(
            tenant_id="tenantId",
            grant_id="grantId",
        )
        """
        _response = self._raw_client.get(tenant_id, grant_id, request_options=request_options)
        return _response.data

    def delete(self, tenant_id: str, grant_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> None:
        """
        Parameters
        ----------
        tenant_id : str

        grant_id : str

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        None

        Examples
        --------
        from cloudpdf import CloudPDFClient

        client = CloudPDFClient(
            token="YOUR_TOKEN",
        )
        client.grants.delete(
            tenant_id="tenantId",
            grant_id="grantId",
        )
        """
        _response = self._raw_client.delete(tenant_id, grant_id, request_options=request_options)
        return _response.data

    def update(
        self,
        tenant_id: str,
        grant_id: str,
        *,
        name: typing.Optional[str] = OMIT,
        layer_mode: typing.Optional[GrantsUpdateRequestLayerMode] = OMIT,
        layer_name: typing.Optional[str] = OMIT,
        keep_visitor_layers: typing.Optional[bool] = OMIT,
        visitor_idle_seconds: typing.Optional[int] = OMIT,
        visitor_identity: typing.Optional[GrantsUpdateRequestVisitorIdentity] = OMIT,
        identity: typing.Optional[Identity] = OMIT,
        scope: typing.Optional[typing.Sequence[str]] = OMIT,
        origins: typing.Optional[typing.Sequence[str]] = OMIT,
        passphrase: typing.Optional[str] = OMIT,
        session_ttl_seconds: typing.Optional[int] = OMIT,
        disabled: typing.Optional[bool] = OMIT,
        expires_at: typing.Optional[dt.datetime] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> GrantsUpdateResponse:
        """
        Parameters
        ----------
        tenant_id : str

        grant_id : str

        name : typing.Optional[str]

        layer_mode : typing.Optional[GrantsUpdateRequestLayerMode]

        layer_name : typing.Optional[str]

        keep_visitor_layers : typing.Optional[bool]

        visitor_idle_seconds : typing.Optional[int]

        visitor_identity : typing.Optional[GrantsUpdateRequestVisitorIdentity]

        identity : typing.Optional[Identity]

        scope : typing.Optional[typing.Sequence[str]]

        origins : typing.Optional[typing.Sequence[str]]

        passphrase : typing.Optional[str]

        session_ttl_seconds : typing.Optional[int]

        disabled : typing.Optional[bool]

        expires_at : typing.Optional[dt.datetime]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GrantsUpdateResponse
            OK

        Examples
        --------
        from cloudpdf import CloudPDFClient

        client = CloudPDFClient(
            token="YOUR_TOKEN",
        )
        client.grants.update(
            tenant_id="tenantId",
            grant_id="grantId",
        )
        """
        _response = self._raw_client.update(
            tenant_id,
            grant_id,
            name=name,
            layer_mode=layer_mode,
            layer_name=layer_name,
            keep_visitor_layers=keep_visitor_layers,
            visitor_idle_seconds=visitor_idle_seconds,
            visitor_identity=visitor_identity,
            identity=identity,
            scope=scope,
            origins=origins,
            passphrase=passphrase,
            session_ttl_seconds=session_ttl_seconds,
            disabled=disabled,
            expires_at=expires_at,
            request_options=request_options,
        )
        return _response.data

    def archive(
        self, tenant_id: str, grant_id: str, *, request_options: typing.Optional[RequestOptions] = None
    ) -> GrantsArchiveResponse:
        """
        Archiving is final: an archived grant never exchanges again, and to give access again you create a new grant. Pausing (`disabled`) is the reversible stop. An archived grant no longer counts as using its shared layer, so that layer can be deleted.

        Parameters
        ----------
        tenant_id : str

        grant_id : str

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GrantsArchiveResponse
            OK

        Examples
        --------
        from cloudpdf import CloudPDFClient

        client = CloudPDFClient(
            token="YOUR_TOKEN",
        )
        client.grants.archive(
            tenant_id="tenantId",
            grant_id="grantId",
        )
        """
        _response = self._raw_client.archive(tenant_id, grant_id, request_options=request_options)
        return _response.data

    @property
    def visitors(self):
        if self._visitors is None:
            from .visitors.client import VisitorsClient  # noqa: E402

            self._visitors = VisitorsClient(client_wrapper=self._client_wrapper)
        return self._visitors


class AsyncGrantsClient:
    def __init__(self, *, client_wrapper: AsyncClientWrapper):
        self._raw_client = AsyncRawGrantsClient(client_wrapper=client_wrapper)
        self._client_wrapper = client_wrapper
        self._visitors: typing.Optional[AsyncVisitorsClient] = None

    @property
    def with_raw_response(self) -> AsyncRawGrantsClient:
        """
        Retrieves a raw implementation of this client that returns raw responses.

        Returns
        -------
        AsyncRawGrantsClient
        """
        return self._raw_client

    async def exchange(
        self,
        *,
        grant: str,
        passphrase: typing.Optional[str] = OMIT,
        visitor_handle: typing.Optional[str] = OMIT,
        restart: typing.Optional[bool] = OMIT,
        visitor_identity: typing.Optional[GrantsExchangeRequestVisitorIdentity] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> GrantsExchangeResponse:
        """
        Unauthenticated, but requires a browser Origin header, checked against the grant allowlist. Unknown, deleted, paused and archived grants are indistinguishable (404). Passphrase-protected grants return 422 GrantPassphraseRequired until `passphrase` is supplied. A grant that asks its visitors who they are on open returns 422 VisitorIdentityRequired until the required answers are given; `error.details` carries `fields` (what the grant asks), `missing`, `invalid`, `purpose` and `privacyUrl`. A grant that asks before the first change issues the session at once with `visitor.identityRequest`, and writes answer 403 VisitorIdentityRequired with the same details until the visitor exchanges again with their answers. Answers that can not be taken answer 422 on any grant. Mounted only when the deployment can sign (HS256 mode).

        Parameters
        ----------
        grant : str

        passphrase : typing.Optional[str]

        visitor_handle : typing.Optional[str]

        restart : typing.Optional[bool]

        visitor_identity : typing.Optional[GrantsExchangeRequestVisitorIdentity]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GrantsExchangeResponse
            OK

        Examples
        --------
        import asyncio

        from cloudpdf import AsyncCloudPDFClient

        client = AsyncCloudPDFClient(
            token="YOUR_TOKEN",
        )


        async def main() -> None:
            await client.grants.exchange(
                grant="grant",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.exchange(
            grant=grant,
            passphrase=passphrase,
            visitor_handle=visitor_handle,
            restart=restart,
            visitor_identity=visitor_identity,
            request_options=request_options,
        )
        return _response.data

    async def list(
        self,
        tenant_id: str,
        *,
        limit: typing.Optional[int] = None,
        cursor: typing.Optional[str] = None,
        doc_id: typing.Optional[str] = None,
        archived: typing.Optional[bool] = None,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> GrantsListResponse:
        """
        Parameters
        ----------
        tenant_id : str

        limit : typing.Optional[int]

        cursor : typing.Optional[str]

        doc_id : typing.Optional[str]

        archived : typing.Optional[bool]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GrantsListResponse
            OK

        Examples
        --------
        import asyncio

        from cloudpdf import AsyncCloudPDFClient

        client = AsyncCloudPDFClient(
            token="YOUR_TOKEN",
        )


        async def main() -> None:
            await client.grants.list(
                tenant_id="tenantId",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.list(
            tenant_id, limit=limit, cursor=cursor, doc_id=doc_id, archived=archived, request_options=request_options
        )
        return _response.data

    async def create(
        self,
        tenant_id: str,
        *,
        doc_id: str,
        scope: typing.Sequence[str],
        name: typing.Optional[str] = OMIT,
        layer_mode: typing.Optional[GrantsCreateRequestLayerMode] = OMIT,
        layer_name: typing.Optional[str] = OMIT,
        keep_visitor_layers: typing.Optional[bool] = OMIT,
        visitor_idle_seconds: typing.Optional[int] = OMIT,
        visitor_identity: typing.Optional[GrantsCreateRequestVisitorIdentity] = OMIT,
        identity: typing.Optional[Identity] = OMIT,
        origins: typing.Optional[typing.Sequence[str]] = OMIT,
        passphrase: typing.Optional[str] = OMIT,
        session_ttl_seconds: typing.Optional[int] = OMIT,
        expires_at: typing.Optional[dt.datetime] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> GrantsCreateResponse:
        """
        The returned id is what a link page or an embed opens. Its scope gives only document permissions the caller holds itself, as tokens.issue does. Mounted only when the deployment can sign (HS256 mode) — exchange mints session JWTs, so grants exist only where minting does.

        Parameters
        ----------
        tenant_id : str

        doc_id : str

        scope : typing.Sequence[str]

        name : typing.Optional[str]

        layer_mode : typing.Optional[GrantsCreateRequestLayerMode]

        layer_name : typing.Optional[str]

        keep_visitor_layers : typing.Optional[bool]

        visitor_idle_seconds : typing.Optional[int]

        visitor_identity : typing.Optional[GrantsCreateRequestVisitorIdentity]

        identity : typing.Optional[Identity]

        origins : typing.Optional[typing.Sequence[str]]

        passphrase : typing.Optional[str]

        session_ttl_seconds : typing.Optional[int]

        expires_at : typing.Optional[dt.datetime]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GrantsCreateResponse
            OK

        Examples
        --------
        import asyncio

        from cloudpdf import AsyncCloudPDFClient

        client = AsyncCloudPDFClient(
            token="YOUR_TOKEN",
        )


        async def main() -> None:
            await client.grants.create(
                tenant_id="tenantId",
                doc_id="docId",
                scope=["scope"],
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.create(
            tenant_id,
            doc_id=doc_id,
            scope=scope,
            name=name,
            layer_mode=layer_mode,
            layer_name=layer_name,
            keep_visitor_layers=keep_visitor_layers,
            visitor_idle_seconds=visitor_idle_seconds,
            visitor_identity=visitor_identity,
            identity=identity,
            origins=origins,
            passphrase=passphrase,
            session_ttl_seconds=session_ttl_seconds,
            expires_at=expires_at,
            request_options=request_options,
        )
        return _response.data

    async def get(
        self, tenant_id: str, grant_id: str, *, request_options: typing.Optional[RequestOptions] = None
    ) -> GrantsGetResponse:
        """
        Parameters
        ----------
        tenant_id : str

        grant_id : str

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GrantsGetResponse
            OK

        Examples
        --------
        import asyncio

        from cloudpdf import AsyncCloudPDFClient

        client = AsyncCloudPDFClient(
            token="YOUR_TOKEN",
        )


        async def main() -> None:
            await client.grants.get(
                tenant_id="tenantId",
                grant_id="grantId",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.get(tenant_id, grant_id, request_options=request_options)
        return _response.data

    async def delete(
        self, tenant_id: str, grant_id: str, *, request_options: typing.Optional[RequestOptions] = None
    ) -> None:
        """
        Parameters
        ----------
        tenant_id : str

        grant_id : str

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        None

        Examples
        --------
        import asyncio

        from cloudpdf import AsyncCloudPDFClient

        client = AsyncCloudPDFClient(
            token="YOUR_TOKEN",
        )


        async def main() -> None:
            await client.grants.delete(
                tenant_id="tenantId",
                grant_id="grantId",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.delete(tenant_id, grant_id, request_options=request_options)
        return _response.data

    async def update(
        self,
        tenant_id: str,
        grant_id: str,
        *,
        name: typing.Optional[str] = OMIT,
        layer_mode: typing.Optional[GrantsUpdateRequestLayerMode] = OMIT,
        layer_name: typing.Optional[str] = OMIT,
        keep_visitor_layers: typing.Optional[bool] = OMIT,
        visitor_idle_seconds: typing.Optional[int] = OMIT,
        visitor_identity: typing.Optional[GrantsUpdateRequestVisitorIdentity] = OMIT,
        identity: typing.Optional[Identity] = OMIT,
        scope: typing.Optional[typing.Sequence[str]] = OMIT,
        origins: typing.Optional[typing.Sequence[str]] = OMIT,
        passphrase: typing.Optional[str] = OMIT,
        session_ttl_seconds: typing.Optional[int] = OMIT,
        disabled: typing.Optional[bool] = OMIT,
        expires_at: typing.Optional[dt.datetime] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> GrantsUpdateResponse:
        """
        Parameters
        ----------
        tenant_id : str

        grant_id : str

        name : typing.Optional[str]

        layer_mode : typing.Optional[GrantsUpdateRequestLayerMode]

        layer_name : typing.Optional[str]

        keep_visitor_layers : typing.Optional[bool]

        visitor_idle_seconds : typing.Optional[int]

        visitor_identity : typing.Optional[GrantsUpdateRequestVisitorIdentity]

        identity : typing.Optional[Identity]

        scope : typing.Optional[typing.Sequence[str]]

        origins : typing.Optional[typing.Sequence[str]]

        passphrase : typing.Optional[str]

        session_ttl_seconds : typing.Optional[int]

        disabled : typing.Optional[bool]

        expires_at : typing.Optional[dt.datetime]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GrantsUpdateResponse
            OK

        Examples
        --------
        import asyncio

        from cloudpdf import AsyncCloudPDFClient

        client = AsyncCloudPDFClient(
            token="YOUR_TOKEN",
        )


        async def main() -> None:
            await client.grants.update(
                tenant_id="tenantId",
                grant_id="grantId",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.update(
            tenant_id,
            grant_id,
            name=name,
            layer_mode=layer_mode,
            layer_name=layer_name,
            keep_visitor_layers=keep_visitor_layers,
            visitor_idle_seconds=visitor_idle_seconds,
            visitor_identity=visitor_identity,
            identity=identity,
            scope=scope,
            origins=origins,
            passphrase=passphrase,
            session_ttl_seconds=session_ttl_seconds,
            disabled=disabled,
            expires_at=expires_at,
            request_options=request_options,
        )
        return _response.data

    async def archive(
        self, tenant_id: str, grant_id: str, *, request_options: typing.Optional[RequestOptions] = None
    ) -> GrantsArchiveResponse:
        """
        Archiving is final: an archived grant never exchanges again, and to give access again you create a new grant. Pausing (`disabled`) is the reversible stop. An archived grant no longer counts as using its shared layer, so that layer can be deleted.

        Parameters
        ----------
        tenant_id : str

        grant_id : str

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        GrantsArchiveResponse
            OK

        Examples
        --------
        import asyncio

        from cloudpdf import AsyncCloudPDFClient

        client = AsyncCloudPDFClient(
            token="YOUR_TOKEN",
        )


        async def main() -> None:
            await client.grants.archive(
                tenant_id="tenantId",
                grant_id="grantId",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.archive(tenant_id, grant_id, request_options=request_options)
        return _response.data

    @property
    def visitors(self):
        if self._visitors is None:
            from .visitors.client import AsyncVisitorsClient  # noqa: E402

            self._visitors = AsyncVisitorsClient(client_wrapper=self._client_wrapper)
        return self._visitors