# This file is auto-generated by oagen. Do not edit.

from __future__ import annotations

from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from .._client import AsyncWorkOSClient, WorkOSClient

from workos.common.models.connected_account import ConnectedAccount
from workos.common.models.connected_account_input_state import (
    ConnectedAccountInputState,
)
from workos.common.models.create_data_integration_auth_methods import (
    CreateDataIntegrationAuthMethods,
)
from workos.common.models.pagination_order import PaginationOrder

from .._pagination import AsyncPage, SyncPage
from .._types import NOT_GIVEN, NotGiven, RequestOptions, enum_value
from .models import (
    ApiKeyInstallation,
    CustomProviderDefinition,
    DataIntegration,
    DataIntegrationAccessTokenResponse,
    DataIntegrationAuthorizeUrlResponse,
    DataIntegrationCredentialsInput,
    DataIntegrationCredentialsResponse,
    DataIntegrationsListResponse,
    UpdateCustomProviderDefinition,
)


class Pipes:
    """Pipes API resources."""

    def __init__(self, client: WorkOSClient) -> None:
        self._client = client

    def list_data_integrations(
        self,
        *,
        limit: int | None = None,
        before: str | None = None,
        after: str | None = None,
        order: PaginationOrder | str | None = "desc",
        request_options: RequestOptions | None = None,
    ) -> SyncPage[DataIntegration]:
        """List data integrations

        Lists the environment's data integrations configured with `custom` or `organization` credentials, including custom providers and API key integrations.

        Args:
            limit: Upper limit on the number of objects to return, between `1` and `100`. Defaults to `10`.
            before: An object ID that defines your place in the list. When the ID is not present, you are at the end of the list. For example, if you make a list request and receive 100 objects, ending with `"obj_123"`, your subsequent call can include `before="obj_123"` to fetch a new batch of objects before `"obj_123"`.
            after: An object ID that defines your place in the list. When the ID is not present, you are at the end of the list. For example, if you make a list request and receive 100 objects, ending with `"obj_123"`, your subsequent call can include `after="obj_123"` to fetch a new batch of objects after `"obj_123"`.
            order: Order the results by the creation time. Supported values are `"asc"` (ascending), `"desc"` (descending), and `"normal"` (descending with reversed cursor semantics where `before` fetches older records and `after` fetches newer records). Defaults to `desc`.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            SyncPage[DataIntegration]

        Raises:
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params = {
            k: v
            for k, v in {
                "limit": limit,
                "before": before,
                "after": after,
                "order": enum_value(order) if order is not None else None,
            }.items()
            if v is not None
        }
        return self._client.request_page(
            method="get",
            path=("data-integrations",),
            model=DataIntegration,
            params=params,
            request_options=request_options,
        )

    def create_data_integration(
        self,
        *,
        provider: str,
        description: str | None | NotGiven = NOT_GIVEN,
        enabled: bool | None = None,
        scopes: list[str] | None | NotGiven = NOT_GIVEN,
        auth_methods: list[CreateDataIntegrationAuthMethods | str] | None = None,
        config: dict[str, str] | None = None,
        credentials: DataIntegrationCredentialsInput | None = None,
        api_key: ApiKeyInstallation | None = None,
        custom_provider: CustomProviderDefinition | None = None,
        request_options: RequestOptions | None = None,
    ) -> DataIntegration:
        """Create a data integration

        Creates a data integration for a provider. Set `credentials.type` to `custom` to use your own OAuth app credentials or `organization` to have each organization supply its own. Set `auth_methods` to `["api_key"]` to create an API key integration; you may optionally supply an `api_key` block to install a first tenant in the same call. For a built-in provider, pass its slug as `provider`. For a custom provider, pass a new slug plus a `custom_provider` definition.

        Args:
            provider: The provider to create a Data Integration for. For a built-in provider use its slug (e.g. `github`, `slack`). For a custom provider, this is the new provider slug and `custom_provider` must be supplied. A custom provider slug cannot shadow an existing global provider slug.
            description: An optional description of the Data Integration.
            enabled: Whether the Data Integration is enabled. Defaults to `false`.
            scopes: The OAuth scopes to request for the Data Integration. Defaults to the provider's configured scopes when omitted.
            auth_methods: How accounts authenticate with the provider. Defaults to `["oauth"]`. Use `["api_key"]` to declare an API key integration; `credentials` is then not required and keys are supplied per-tenant (optionally via `api_key` on this request).
            config: Provider-specific config values (e.g. a Snowflake `account_identifier`), keyed by the config field. Only fields the built-in provider declares are accepted.
            credentials: The OAuth credentials to configure for the Data Integration. Required for OAuth integrations; omit when `auth_methods` is `["api_key"]`.
            api_key: An optional API key to install for the first tenant on an `api_key` integration. Omit to declare a keyless integration; tenants can be added later via the per-installation API key path.
            custom_provider: The OAuth definition for a custom provider. Supply this to define a custom provider; omit it to create an integration for a built-in provider.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DataIntegration

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            AuthorizationError: If the request is forbidden (403).
            NotFoundError: If the resource is not found (404).
            ConflictError: If a conflict occurs (409).
            UnprocessableEntityError: If the request data is unprocessable (422).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            k: v
            for k, v in {
                "provider": provider,
                "enabled": enabled,
                "auth_methods": auth_methods,
                "config": config,
                "credentials": credentials.to_dict()
                if credentials is not None
                else None,
                "api_key": api_key.to_dict() if api_key is not None else None,
                "custom_provider": custom_provider.to_dict()
                if custom_provider is not None
                else None,
            }.items()
            if v is not None
        }
        if not isinstance(description, NotGiven):
            body["description"] = description
        if not isinstance(scopes, NotGiven):
            body["scopes"] = scopes
        return self._client.request(
            method="post",
            path=("data-integrations",),
            body=body,
            model=DataIntegration,
            request_options=request_options,
        )

    def get_data_integration(
        self,
        slug: str,
        *,
        request_options: RequestOptions | None = None,
    ) -> DataIntegration:
        """Get a data integration

        Retrieves a data integration by its slug.

        Args:
            slug: The slug identifier of the data integration.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DataIntegration

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        return self._client.request(
            method="get",
            path=("data-integrations", str(slug)),
            model=DataIntegration,
            request_options=request_options,
        )

    def update_data_integration(
        self,
        slug: str,
        *,
        description: str | None | NotGiven = NOT_GIVEN,
        enabled: bool | None = None,
        scopes: list[str] | None | NotGiven = NOT_GIVEN,
        credentials: DataIntegrationCredentialsInput | None = None,
        api_key: ApiKeyInstallation | None = None,
        custom_provider: UpdateCustomProviderDefinition | None = None,
        request_options: RequestOptions | None = None,
    ) -> DataIntegration:
        """Update a data integration

        Updates the description, enabled state, or custom credentials of a data integration. For custom providers, `custom_provider` updates the OAuth definition.

        Args:
            slug: The slug identifier of the data integration.
            description: An optional description of the Data Integration.
            enabled: Whether the Data Integration is enabled.
            scopes: The OAuth scopes to request for the Data Integration. Pass `null` to reset to the provider's configured scopes.
            credentials: New OAuth credentials for the Data Integration. When provided, rotates the stored client secret. Mutually exclusive with `api_key`.
            api_key: An API key to install or rotate for a tenant on an `api_key` integration. Upserts the tenant installation identified by `user_id` (and optional `organization_id`).
            custom_provider: Updates to a custom provider's OAuth definition. Only valid for custom-provider integrations.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DataIntegration

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            AuthorizationError: If the request is forbidden (403).
            NotFoundError: If the resource is not found (404).
            UnprocessableEntityError: If the request data is unprocessable (422).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            k: v
            for k, v in {
                "enabled": enabled,
                "credentials": credentials.to_dict()
                if credentials is not None
                else None,
                "api_key": api_key.to_dict() if api_key is not None else None,
                "custom_provider": custom_provider.to_dict()
                if custom_provider is not None
                else None,
            }.items()
            if v is not None
        }
        if not isinstance(description, NotGiven):
            body["description"] = description
        if not isinstance(scopes, NotGiven):
            body["scopes"] = scopes
        return self._client.request(
            method="put",
            path=("data-integrations", str(slug)),
            body=body,
            model=DataIntegration,
            request_options=request_options,
        )

    def delete_data_integration(
        self,
        slug: str,
        *,
        request_options: RequestOptions | None = None,
    ) -> None:
        """Delete a data integration

        Deletes a data integration and all of its connected installations. For a custom provider, also deletes the custom provider definition.

        Args:
            slug: The slug identifier of the data integration.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        self._client.request(
            method="delete",
            path=("data-integrations", str(slug)),
            request_options=request_options,
        )

    def update_data_integration_api_key(
        self,
        slug: str,
        *,
        user_id: str,
        secret: str,
        organization_id: str | None = None,
        request_options: RequestOptions | None = None,
    ) -> ConnectedAccount:
        """Upsert an API key for a connected account

        Creates or updates an API-key-based installation for the specified integration and user. If an installation already exists, the stored API key is rotated to the new value.

        Args:
            slug: The identifier of the integration.
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier.
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter to scope the connection to a specific organization.
            secret: The API key secret to store for this integration.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            ConnectedAccount

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            AuthorizationError: If the request is forbidden (403).
            NotFoundError: If the resource is not found (404).
            UnprocessableEntityError: If the request data is unprocessable (422).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            k: v
            for k, v in {
                "user_id": user_id,
                "organization_id": organization_id,
                "secret": secret,
            }.items()
            if v is not None
        }
        return self._client.request(
            method="put",
            path=("data-integrations", str(slug), "api-key"),
            body=body,
            model=ConnectedAccount,
            request_options=request_options,
        )

    def authorize_data_integration(
        self,
        slug: str,
        *,
        user_id: str,
        organization_id: str | None = None,
        return_to: str | None = None,
        config: dict[str, str] | None = None,
        request_options: RequestOptions | None = None,
    ) -> DataIntegrationAuthorizeUrlResponse:
        """Get authorization URL

        Generates an OAuth authorization URL to initiate the connection flow for a user. Redirect the user to the returned URL to begin the OAuth flow with the third-party provider.

        Args:
            slug: The slug identifier of the provider (e.g., `github`, `slack`, `notion`).
            user_id: The ID of the user to authorize.
            organization_id: An organization ID to scope the authorization to a specific organization.
            return_to: The URL to redirect the user to after authorization.
            config: Connect-time config values for the provider-declared `installation`-scope fields (e.g. a Zendesk `subdomain`), keyed by the config field. Only fields the provider declares may be supplied, and required fields must be provided unless already pinned on the integration.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DataIntegrationAuthorizeUrlResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            AuthorizationError: If the request is forbidden (403).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            k: v
            for k, v in {
                "user_id": user_id,
                "organization_id": organization_id,
                "return_to": return_to,
                "config": config,
            }.items()
            if v is not None
        }
        return self._client.request(
            method="post",
            path=("data-integrations", str(slug), "authorize"),
            body=body,
            model=DataIntegrationAuthorizeUrlResponse,
            request_options=request_options,
        )

    def create_data_integration_credential(
        self,
        slug: str,
        *,
        user_id: str,
        organization_id: str | None = None,
        request_options: RequestOptions | None = None,
    ) -> DataIntegrationCredentialsResponse:
        """Vend credentials for a connected account

        Returns credentials for a user's connected account. Branches on the installation's `auth_method`: OAuth installations return an access token (refreshed if needed); API-key installations return the stored secret.

        Args:
            slug: The identifier of the integration.
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier.
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter to scope the connection to a specific organization.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DataIntegrationCredentialsResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            k: v
            for k, v in {
                "user_id": user_id,
                "organization_id": organization_id,
            }.items()
            if v is not None
        }
        return self._client.request(
            method="post",
            path=("data-integrations", str(slug), "credentials"),
            body=body,
            model=DataIntegrationCredentialsResponse,
            request_options=request_options,
        )

    def get_access_token(
        self,
        provider: str,
        *,
        user_id: str,
        organization_id: str | None | NotGiven = NOT_GIVEN,
        request_options: RequestOptions | None = None,
    ) -> DataIntegrationAccessTokenResponse:
        """Get an access token for a connected account

        Fetches a valid OAuth access token for a user's connected account. WorkOS automatically handles token refresh, ensuring you always receive a valid, non-expired token.

        Args:
            provider: The identifier of the integration.
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier.
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter to scope the connection to a specific organization.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DataIntegrationAccessTokenResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            UnprocessableEntityError: If the request data is unprocessable (422).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            "user_id": user_id,
        }
        if not isinstance(organization_id, NotGiven):
            body["organization_id"] = organization_id
        return self._client.request(
            method="post",
            path=("data-integrations", str(provider), "token"),
            body=body,
            model=DataIntegrationAccessTokenResponse,
            request_options=request_options,
        )

    def get_user_connected_account(
        self,
        user_id: str,
        slug: str,
        *,
        organization_id: str | None = None,
        request_options: RequestOptions | None = None,
    ) -> ConnectedAccount:
        """Get a connected account

        Retrieves a user's [connected account](https://workos.com/docs/reference/pipes/connected-account) for a specific provider.

        Args:
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier.
            slug: The slug identifier of the provider (e.g., `github`, `slack`, `notion`).
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter if the connection is scoped to an organization.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            ConnectedAccount

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params: dict[str, Any] = {
            k: v
            for k, v in {
                "organization_id": organization_id,
            }.items()
            if v is not None
        }
        return self._client.request(
            method="get",
            path=(
                "user_management",
                "users",
                str(user_id),
                "connected_accounts",
                str(slug),
            ),
            params=params,
            model=ConnectedAccount,
            request_options=request_options,
        )

    def create_user_connected_account(
        self,
        user_id: str,
        slug: str,
        *,
        access_token: str | None = None,
        refresh_token: str | None = None,
        expires_at: str | None = None,
        scopes: list[str] | None = None,
        state: ConnectedAccountInputState | str | None = None,
        organization_id: str | None = None,
        request_options: RequestOptions | None = None,
    ) -> ConnectedAccount:
        """Import a connected account

        Imports a [connected account](https://workos.com/docs/reference/pipes/connected-account) for a user by providing OAuth tokens directly. Use this to migrate existing connections or set up connections without going through the OAuth flow.

        Args:
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier.
            slug: The slug identifier of the provider (e.g., `github`, `slack`, `notion`).
            access_token: The OAuth access token for the connected account.
            refresh_token: The OAuth refresh token for the connected account.
            expires_at: The ISO-8601 timestamp when the access token expires. Required when `access_token` is provided for tokens that expire.
            scopes: The OAuth scopes granted for this connection.
            state: Explicitly set the state of the connected account. When omitted, the state is derived from the token combination provided.
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter if the connection is scoped to an organization.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            ConnectedAccount

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            ConflictError: If a conflict occurs (409).
            UnprocessableEntityError: If the request data is unprocessable (422).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            k: v
            for k, v in {
                "access_token": access_token,
                "refresh_token": refresh_token,
                "expires_at": expires_at,
                "scopes": scopes,
                "state": enum_value(state) if state is not None else None,
            }.items()
            if v is not None
        }
        params: dict[str, Any] = {
            k: v
            for k, v in {
                "organization_id": organization_id,
            }.items()
            if v is not None
        }
        return self._client.request(
            method="post",
            path=(
                "user_management",
                "users",
                str(user_id),
                "connected_accounts",
                str(slug),
            ),
            body=body,
            params=params,
            model=ConnectedAccount,
            request_options=request_options,
        )

    def update_user_connected_account(
        self,
        user_id: str,
        slug: str,
        *,
        access_token: str | None = None,
        refresh_token: str | None = None,
        expires_at: str | None = None,
        scopes: list[str] | None = None,
        state: ConnectedAccountInputState | str | None = None,
        organization_id: str | None = None,
        request_options: RequestOptions | None = None,
    ) -> ConnectedAccount:
        """Update a connected account

        Updates a user's [connected account](https://workos.com/docs/reference/pipes/connected-account) tokens, scopes, or state for a specific provider.

        Args:
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier.
            slug: The slug identifier of the provider (e.g., `github`, `slack`, `notion`).
            access_token: The OAuth access token for the connected account.
            refresh_token: The OAuth refresh token for the connected account.
            expires_at: The ISO-8601 timestamp when the access token expires. Required when `access_token` is provided for tokens that expire.
            scopes: The OAuth scopes granted for this connection.
            state: Explicitly set the state of the connected account. When omitted, the state is derived from the token combination provided.
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter if the connection is scoped to an organization.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            ConnectedAccount

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            k: v
            for k, v in {
                "access_token": access_token,
                "refresh_token": refresh_token,
                "expires_at": expires_at,
                "scopes": scopes,
                "state": enum_value(state) if state is not None else None,
            }.items()
            if v is not None
        }
        params: dict[str, Any] = {
            k: v
            for k, v in {
                "organization_id": organization_id,
            }.items()
            if v is not None
        }
        return self._client.request(
            method="put",
            path=(
                "user_management",
                "users",
                str(user_id),
                "connected_accounts",
                str(slug),
            ),
            body=body,
            params=params,
            model=ConnectedAccount,
            request_options=request_options,
        )

    def delete_user_connected_account(
        self,
        user_id: str,
        slug: str,
        *,
        organization_id: str | None = None,
        request_options: RequestOptions | None = None,
    ) -> None:
        """Delete a connected account

        Disconnects WorkOS's account for the user, including removing any stored access and refresh tokens. The user will need to reauthorize if they want to reconnect. This does not revoke access on the provider side.

        Args:
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier.
            slug: The slug identifier of the provider (e.g., `github`, `slack`, `notion`).
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter if the connection is scoped to an organization.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params: dict[str, Any] = {
            k: v
            for k, v in {
                "organization_id": organization_id,
            }.items()
            if v is not None
        }
        self._client.request(
            method="delete",
            path=(
                "user_management",
                "users",
                str(user_id),
                "connected_accounts",
                str(slug),
            ),
            params=params,
            request_options=request_options,
        )

    def list_user_data_providers(
        self,
        user_id: str,
        *,
        organization_id: str | None = None,
        request_options: RequestOptions | None = None,
    ) -> DataIntegrationsListResponse:
        """List providers for a user

        Retrieves a list of available providers and the user's connection status for each. Returns all providers configured for your environment, along with the user's [connected account](https://workos.com/docs/reference/pipes/connected-account) information where applicable.

        Args:
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier to list providers and connected accounts for.
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter to filter connections for a specific organization.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DataIntegrationsListResponse

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params: dict[str, Any] = {
            k: v
            for k, v in {
                "organization_id": organization_id,
            }.items()
            if v is not None
        }
        return self._client.request(
            method="get",
            path=("user_management", "users", str(user_id), "data_providers"),
            params=params,
            model=DataIntegrationsListResponse,
            request_options=request_options,
        )


class AsyncPipes:
    """Pipes API resources (async)."""

    def __init__(self, client: AsyncWorkOSClient) -> None:
        self._client = client

    async def list_data_integrations(
        self,
        *,
        limit: int | None = None,
        before: str | None = None,
        after: str | None = None,
        order: PaginationOrder | str | None = "desc",
        request_options: RequestOptions | None = None,
    ) -> AsyncPage[DataIntegration]:
        """List data integrations

        Lists the environment's data integrations configured with `custom` or `organization` credentials, including custom providers and API key integrations.

        Args:
            limit: Upper limit on the number of objects to return, between `1` and `100`. Defaults to `10`.
            before: An object ID that defines your place in the list. When the ID is not present, you are at the end of the list. For example, if you make a list request and receive 100 objects, ending with `"obj_123"`, your subsequent call can include `before="obj_123"` to fetch a new batch of objects before `"obj_123"`.
            after: An object ID that defines your place in the list. When the ID is not present, you are at the end of the list. For example, if you make a list request and receive 100 objects, ending with `"obj_123"`, your subsequent call can include `after="obj_123"` to fetch a new batch of objects after `"obj_123"`.
            order: Order the results by the creation time. Supported values are `"asc"` (ascending), `"desc"` (descending), and `"normal"` (descending with reversed cursor semantics where `before` fetches older records and `after` fetches newer records). Defaults to `desc`.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            AsyncPage[DataIntegration]

        Raises:
            AuthenticationError: If the API key is invalid (401).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params = {
            k: v
            for k, v in {
                "limit": limit,
                "before": before,
                "after": after,
                "order": enum_value(order) if order is not None else None,
            }.items()
            if v is not None
        }
        return await self._client.request_page(
            method="get",
            path=("data-integrations",),
            model=DataIntegration,
            params=params,
            request_options=request_options,
        )

    async def create_data_integration(
        self,
        *,
        provider: str,
        description: str | None | NotGiven = NOT_GIVEN,
        enabled: bool | None = None,
        scopes: list[str] | None | NotGiven = NOT_GIVEN,
        auth_methods: list[CreateDataIntegrationAuthMethods | str] | None = None,
        config: dict[str, str] | None = None,
        credentials: DataIntegrationCredentialsInput | None = None,
        api_key: ApiKeyInstallation | None = None,
        custom_provider: CustomProviderDefinition | None = None,
        request_options: RequestOptions | None = None,
    ) -> DataIntegration:
        """Create a data integration

        Creates a data integration for a provider. Set `credentials.type` to `custom` to use your own OAuth app credentials or `organization` to have each organization supply its own. Set `auth_methods` to `["api_key"]` to create an API key integration; you may optionally supply an `api_key` block to install a first tenant in the same call. For a built-in provider, pass its slug as `provider`. For a custom provider, pass a new slug plus a `custom_provider` definition.

        Args:
            provider: The provider to create a Data Integration for. For a built-in provider use its slug (e.g. `github`, `slack`). For a custom provider, this is the new provider slug and `custom_provider` must be supplied. A custom provider slug cannot shadow an existing global provider slug.
            description: An optional description of the Data Integration.
            enabled: Whether the Data Integration is enabled. Defaults to `false`.
            scopes: The OAuth scopes to request for the Data Integration. Defaults to the provider's configured scopes when omitted.
            auth_methods: How accounts authenticate with the provider. Defaults to `["oauth"]`. Use `["api_key"]` to declare an API key integration; `credentials` is then not required and keys are supplied per-tenant (optionally via `api_key` on this request).
            config: Provider-specific config values (e.g. a Snowflake `account_identifier`), keyed by the config field. Only fields the built-in provider declares are accepted.
            credentials: The OAuth credentials to configure for the Data Integration. Required for OAuth integrations; omit when `auth_methods` is `["api_key"]`.
            api_key: An optional API key to install for the first tenant on an `api_key` integration. Omit to declare a keyless integration; tenants can be added later via the per-installation API key path.
            custom_provider: The OAuth definition for a custom provider. Supply this to define a custom provider; omit it to create an integration for a built-in provider.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DataIntegration

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            AuthorizationError: If the request is forbidden (403).
            NotFoundError: If the resource is not found (404).
            ConflictError: If a conflict occurs (409).
            UnprocessableEntityError: If the request data is unprocessable (422).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            k: v
            for k, v in {
                "provider": provider,
                "enabled": enabled,
                "auth_methods": auth_methods,
                "config": config,
                "credentials": credentials.to_dict()
                if credentials is not None
                else None,
                "api_key": api_key.to_dict() if api_key is not None else None,
                "custom_provider": custom_provider.to_dict()
                if custom_provider is not None
                else None,
            }.items()
            if v is not None
        }
        if not isinstance(description, NotGiven):
            body["description"] = description
        if not isinstance(scopes, NotGiven):
            body["scopes"] = scopes
        return await self._client.request(
            method="post",
            path=("data-integrations",),
            body=body,
            model=DataIntegration,
            request_options=request_options,
        )

    async def get_data_integration(
        self,
        slug: str,
        *,
        request_options: RequestOptions | None = None,
    ) -> DataIntegration:
        """Get a data integration

        Retrieves a data integration by its slug.

        Args:
            slug: The slug identifier of the data integration.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DataIntegration

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        return await self._client.request(
            method="get",
            path=("data-integrations", str(slug)),
            model=DataIntegration,
            request_options=request_options,
        )

    async def update_data_integration(
        self,
        slug: str,
        *,
        description: str | None | NotGiven = NOT_GIVEN,
        enabled: bool | None = None,
        scopes: list[str] | None | NotGiven = NOT_GIVEN,
        credentials: DataIntegrationCredentialsInput | None = None,
        api_key: ApiKeyInstallation | None = None,
        custom_provider: UpdateCustomProviderDefinition | None = None,
        request_options: RequestOptions | None = None,
    ) -> DataIntegration:
        """Update a data integration

        Updates the description, enabled state, or custom credentials of a data integration. For custom providers, `custom_provider` updates the OAuth definition.

        Args:
            slug: The slug identifier of the data integration.
            description: An optional description of the Data Integration.
            enabled: Whether the Data Integration is enabled.
            scopes: The OAuth scopes to request for the Data Integration. Pass `null` to reset to the provider's configured scopes.
            credentials: New OAuth credentials for the Data Integration. When provided, rotates the stored client secret. Mutually exclusive with `api_key`.
            api_key: An API key to install or rotate for a tenant on an `api_key` integration. Upserts the tenant installation identified by `user_id` (and optional `organization_id`).
            custom_provider: Updates to a custom provider's OAuth definition. Only valid for custom-provider integrations.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DataIntegration

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            AuthorizationError: If the request is forbidden (403).
            NotFoundError: If the resource is not found (404).
            UnprocessableEntityError: If the request data is unprocessable (422).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            k: v
            for k, v in {
                "enabled": enabled,
                "credentials": credentials.to_dict()
                if credentials is not None
                else None,
                "api_key": api_key.to_dict() if api_key is not None else None,
                "custom_provider": custom_provider.to_dict()
                if custom_provider is not None
                else None,
            }.items()
            if v is not None
        }
        if not isinstance(description, NotGiven):
            body["description"] = description
        if not isinstance(scopes, NotGiven):
            body["scopes"] = scopes
        return await self._client.request(
            method="put",
            path=("data-integrations", str(slug)),
            body=body,
            model=DataIntegration,
            request_options=request_options,
        )

    async def delete_data_integration(
        self,
        slug: str,
        *,
        request_options: RequestOptions | None = None,
    ) -> None:
        """Delete a data integration

        Deletes a data integration and all of its connected installations. For a custom provider, also deletes the custom provider definition.

        Args:
            slug: The slug identifier of the data integration.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        await self._client.request(
            method="delete",
            path=("data-integrations", str(slug)),
            request_options=request_options,
        )

    async def update_data_integration_api_key(
        self,
        slug: str,
        *,
        user_id: str,
        secret: str,
        organization_id: str | None = None,
        request_options: RequestOptions | None = None,
    ) -> ConnectedAccount:
        """Upsert an API key for a connected account

        Creates or updates an API-key-based installation for the specified integration and user. If an installation already exists, the stored API key is rotated to the new value.

        Args:
            slug: The identifier of the integration.
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier.
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter to scope the connection to a specific organization.
            secret: The API key secret to store for this integration.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            ConnectedAccount

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            AuthorizationError: If the request is forbidden (403).
            NotFoundError: If the resource is not found (404).
            UnprocessableEntityError: If the request data is unprocessable (422).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            k: v
            for k, v in {
                "user_id": user_id,
                "organization_id": organization_id,
                "secret": secret,
            }.items()
            if v is not None
        }
        return await self._client.request(
            method="put",
            path=("data-integrations", str(slug), "api-key"),
            body=body,
            model=ConnectedAccount,
            request_options=request_options,
        )

    async def authorize_data_integration(
        self,
        slug: str,
        *,
        user_id: str,
        organization_id: str | None = None,
        return_to: str | None = None,
        config: dict[str, str] | None = None,
        request_options: RequestOptions | None = None,
    ) -> DataIntegrationAuthorizeUrlResponse:
        """Get authorization URL

        Generates an OAuth authorization URL to initiate the connection flow for a user. Redirect the user to the returned URL to begin the OAuth flow with the third-party provider.

        Args:
            slug: The slug identifier of the provider (e.g., `github`, `slack`, `notion`).
            user_id: The ID of the user to authorize.
            organization_id: An organization ID to scope the authorization to a specific organization.
            return_to: The URL to redirect the user to after authorization.
            config: Connect-time config values for the provider-declared `installation`-scope fields (e.g. a Zendesk `subdomain`), keyed by the config field. Only fields the provider declares may be supplied, and required fields must be provided unless already pinned on the integration.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DataIntegrationAuthorizeUrlResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            AuthorizationError: If the request is forbidden (403).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            k: v
            for k, v in {
                "user_id": user_id,
                "organization_id": organization_id,
                "return_to": return_to,
                "config": config,
            }.items()
            if v is not None
        }
        return await self._client.request(
            method="post",
            path=("data-integrations", str(slug), "authorize"),
            body=body,
            model=DataIntegrationAuthorizeUrlResponse,
            request_options=request_options,
        )

    async def create_data_integration_credential(
        self,
        slug: str,
        *,
        user_id: str,
        organization_id: str | None = None,
        request_options: RequestOptions | None = None,
    ) -> DataIntegrationCredentialsResponse:
        """Vend credentials for a connected account

        Returns credentials for a user's connected account. Branches on the installation's `auth_method`: OAuth installations return an access token (refreshed if needed); API-key installations return the stored secret.

        Args:
            slug: The identifier of the integration.
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier.
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter to scope the connection to a specific organization.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DataIntegrationCredentialsResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            k: v
            for k, v in {
                "user_id": user_id,
                "organization_id": organization_id,
            }.items()
            if v is not None
        }
        return await self._client.request(
            method="post",
            path=("data-integrations", str(slug), "credentials"),
            body=body,
            model=DataIntegrationCredentialsResponse,
            request_options=request_options,
        )

    async def get_access_token(
        self,
        provider: str,
        *,
        user_id: str,
        organization_id: str | None | NotGiven = NOT_GIVEN,
        request_options: RequestOptions | None = None,
    ) -> DataIntegrationAccessTokenResponse:
        """Get an access token for a connected account

        Fetches a valid OAuth access token for a user's connected account. WorkOS automatically handles token refresh, ensuring you always receive a valid, non-expired token.

        Args:
            provider: The identifier of the integration.
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier.
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter to scope the connection to a specific organization.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DataIntegrationAccessTokenResponse

        Raises:
            BadRequestError: If the request is malformed (400).
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            UnprocessableEntityError: If the request data is unprocessable (422).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            "user_id": user_id,
        }
        if not isinstance(organization_id, NotGiven):
            body["organization_id"] = organization_id
        return await self._client.request(
            method="post",
            path=("data-integrations", str(provider), "token"),
            body=body,
            model=DataIntegrationAccessTokenResponse,
            request_options=request_options,
        )

    async def get_user_connected_account(
        self,
        user_id: str,
        slug: str,
        *,
        organization_id: str | None = None,
        request_options: RequestOptions | None = None,
    ) -> ConnectedAccount:
        """Get a connected account

        Retrieves a user's [connected account](https://workos.com/docs/reference/pipes/connected-account) for a specific provider.

        Args:
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier.
            slug: The slug identifier of the provider (e.g., `github`, `slack`, `notion`).
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter if the connection is scoped to an organization.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            ConnectedAccount

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params: dict[str, Any] = {
            k: v
            for k, v in {
                "organization_id": organization_id,
            }.items()
            if v is not None
        }
        return await self._client.request(
            method="get",
            path=(
                "user_management",
                "users",
                str(user_id),
                "connected_accounts",
                str(slug),
            ),
            params=params,
            model=ConnectedAccount,
            request_options=request_options,
        )

    async def create_user_connected_account(
        self,
        user_id: str,
        slug: str,
        *,
        access_token: str | None = None,
        refresh_token: str | None = None,
        expires_at: str | None = None,
        scopes: list[str] | None = None,
        state: ConnectedAccountInputState | str | None = None,
        organization_id: str | None = None,
        request_options: RequestOptions | None = None,
    ) -> ConnectedAccount:
        """Import a connected account

        Imports a [connected account](https://workos.com/docs/reference/pipes/connected-account) for a user by providing OAuth tokens directly. Use this to migrate existing connections or set up connections without going through the OAuth flow.

        Args:
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier.
            slug: The slug identifier of the provider (e.g., `github`, `slack`, `notion`).
            access_token: The OAuth access token for the connected account.
            refresh_token: The OAuth refresh token for the connected account.
            expires_at: The ISO-8601 timestamp when the access token expires. Required when `access_token` is provided for tokens that expire.
            scopes: The OAuth scopes granted for this connection.
            state: Explicitly set the state of the connected account. When omitted, the state is derived from the token combination provided.
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter if the connection is scoped to an organization.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            ConnectedAccount

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            ConflictError: If a conflict occurs (409).
            UnprocessableEntityError: If the request data is unprocessable (422).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            k: v
            for k, v in {
                "access_token": access_token,
                "refresh_token": refresh_token,
                "expires_at": expires_at,
                "scopes": scopes,
                "state": enum_value(state) if state is not None else None,
            }.items()
            if v is not None
        }
        params: dict[str, Any] = {
            k: v
            for k, v in {
                "organization_id": organization_id,
            }.items()
            if v is not None
        }
        return await self._client.request(
            method="post",
            path=(
                "user_management",
                "users",
                str(user_id),
                "connected_accounts",
                str(slug),
            ),
            body=body,
            params=params,
            model=ConnectedAccount,
            request_options=request_options,
        )

    async def update_user_connected_account(
        self,
        user_id: str,
        slug: str,
        *,
        access_token: str | None = None,
        refresh_token: str | None = None,
        expires_at: str | None = None,
        scopes: list[str] | None = None,
        state: ConnectedAccountInputState | str | None = None,
        organization_id: str | None = None,
        request_options: RequestOptions | None = None,
    ) -> ConnectedAccount:
        """Update a connected account

        Updates a user's [connected account](https://workos.com/docs/reference/pipes/connected-account) tokens, scopes, or state for a specific provider.

        Args:
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier.
            slug: The slug identifier of the provider (e.g., `github`, `slack`, `notion`).
            access_token: The OAuth access token for the connected account.
            refresh_token: The OAuth refresh token for the connected account.
            expires_at: The ISO-8601 timestamp when the access token expires. Required when `access_token` is provided for tokens that expire.
            scopes: The OAuth scopes granted for this connection.
            state: Explicitly set the state of the connected account. When omitted, the state is derived from the token combination provided.
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter if the connection is scoped to an organization.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            ConnectedAccount

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        body: dict[str, Any] = {
            k: v
            for k, v in {
                "access_token": access_token,
                "refresh_token": refresh_token,
                "expires_at": expires_at,
                "scopes": scopes,
                "state": enum_value(state) if state is not None else None,
            }.items()
            if v is not None
        }
        params: dict[str, Any] = {
            k: v
            for k, v in {
                "organization_id": organization_id,
            }.items()
            if v is not None
        }
        return await self._client.request(
            method="put",
            path=(
                "user_management",
                "users",
                str(user_id),
                "connected_accounts",
                str(slug),
            ),
            body=body,
            params=params,
            model=ConnectedAccount,
            request_options=request_options,
        )

    async def delete_user_connected_account(
        self,
        user_id: str,
        slug: str,
        *,
        organization_id: str | None = None,
        request_options: RequestOptions | None = None,
    ) -> None:
        """Delete a connected account

        Disconnects WorkOS's account for the user, including removing any stored access and refresh tokens. The user will need to reauthorize if they want to reconnect. This does not revoke access on the provider side.

        Args:
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier.
            slug: The slug identifier of the provider (e.g., `github`, `slack`, `notion`).
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter if the connection is scoped to an organization.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params: dict[str, Any] = {
            k: v
            for k, v in {
                "organization_id": organization_id,
            }.items()
            if v is not None
        }
        await self._client.request(
            method="delete",
            path=(
                "user_management",
                "users",
                str(user_id),
                "connected_accounts",
                str(slug),
            ),
            params=params,
            request_options=request_options,
        )

    async def list_user_data_providers(
        self,
        user_id: str,
        *,
        organization_id: str | None = None,
        request_options: RequestOptions | None = None,
    ) -> DataIntegrationsListResponse:
        """List providers for a user

        Retrieves a list of available providers and the user's connection status for each. Returns all providers configured for your environment, along with the user's [connected account](https://workos.com/docs/reference/pipes/connected-account) information where applicable.

        Args:
            user_id: A [User](https://workos.com/docs/reference/authkit/user) identifier to list providers and connected accounts for.
            organization_id: An [Organization](https://workos.com/docs/reference/organization) identifier. Optional parameter to filter connections for a specific organization.
            request_options: Per-request options. Supports extra_headers, timeout, max_retries, and base_url override.

        Returns:
            DataIntegrationsListResponse

        Raises:
            AuthenticationError: If the API key is invalid (401).
            NotFoundError: If the resource is not found (404).
            RateLimitExceededError: If rate limited (429).
            ServerError: If the server returns a 5xx error.
        """
        params: dict[str, Any] = {
            k: v
            for k, v in {
                "organization_id": organization_id,
            }.items()
            if v is not None
        }
        return await self._client.request(
            method="get",
            path=("user_management", "users", str(user_id), "data_providers"),
            params=params,
            model=DataIntegrationsListResponse,
            request_options=request_options,
        )
