feat(schema_validation): add mapped endpoints

This commit is contained in:
stainless-app[bot] 2025-05-06 09:48:20 -04:00
parent 1daf1c28c6
commit 7b3a51bc15
38 changed files with 4065 additions and 2 deletions

View file

@ -1,4 +1,4 @@
configured_endpoints: 1714
configured_endpoints: 1727
openapi_spec_url: https://storage.googleapis.com/stainless-sdk-openapi-specs/cloudflare%2Fcloudflare-dacc4cb6d67067b5769858b798c2505c361e4f23167820f52f2543045ee792e8.yml
openapi_spec_hash: b6924a971b9b542de41c687f4add4fe0
config_hash: 98bbabfdd42fef2a932029e37455475f
config_hash: 3c780adf47c2d58e68dc5b7404d7b925

64
api.md
View file

@ -9850,3 +9850,67 @@ Methods:
- <code title="get /accounts/{account_id}/pipelines">client.pipelines.<a href="./src/cloudflare/resources/pipelines.py">list</a>(\*, account_id, \*\*<a href="src/cloudflare/types/pipelines/pipeline_list_params.py">params</a>) -> <a href="./src/cloudflare/types/pipelines/pipeline_list_response.py">PipelineListResponse</a></code>
- <code title="delete /accounts/{account_id}/pipelines/{pipeline_name}">client.pipelines.<a href="./src/cloudflare/resources/pipelines.py">delete</a>(pipeline_name, \*, account_id) -> None</code>
- <code title="get /accounts/{account_id}/pipelines/{pipeline_name}">client.pipelines.<a href="./src/cloudflare/resources/pipelines.py">get</a>(pipeline_name, \*, account_id) -> <a href="./src/cloudflare/types/pipelines/pipeline_get_response.py">PipelineGetResponse</a></code>
# SchemaValidation
## Schemas
Types:
```python
from cloudflare.types.schema_validation import (
SchemaCreateResponse,
SchemaListResponse,
SchemaDeleteResponse,
SchemaEditResponse,
SchemaGetResponse,
)
```
Methods:
- <code title="post /zones/{zone_id}/schema_validation/schemas">client.schema_validation.schemas.<a href="./src/cloudflare/resources/schema_validation/schemas.py">create</a>(\*, zone_id, \*\*<a href="src/cloudflare/types/schema_validation/schema_create_params.py">params</a>) -> <a href="./src/cloudflare/types/schema_validation/schema_create_response.py">SchemaCreateResponse</a></code>
- <code title="get /zones/{zone_id}/schema_validation/schemas">client.schema_validation.schemas.<a href="./src/cloudflare/resources/schema_validation/schemas.py">list</a>(\*, zone_id, \*\*<a href="src/cloudflare/types/schema_validation/schema_list_params.py">params</a>) -> <a href="./src/cloudflare/types/schema_validation/schema_list_response.py">SyncV4PagePaginationArray[SchemaListResponse]</a></code>
- <code title="delete /zones/{zone_id}/schema_validation/schemas/{schema_id}">client.schema_validation.schemas.<a href="./src/cloudflare/resources/schema_validation/schemas.py">delete</a>(schema_id, \*, zone_id) -> <a href="./src/cloudflare/types/schema_validation/schema_delete_response.py">SchemaDeleteResponse</a></code>
- <code title="patch /zones/{zone_id}/schema_validation/schemas/{schema_id}">client.schema_validation.schemas.<a href="./src/cloudflare/resources/schema_validation/schemas.py">edit</a>(schema_id, \*, zone_id, \*\*<a href="src/cloudflare/types/schema_validation/schema_edit_params.py">params</a>) -> <a href="./src/cloudflare/types/schema_validation/schema_edit_response.py">SchemaEditResponse</a></code>
- <code title="get /zones/{zone_id}/schema_validation/schemas/{schema_id}">client.schema_validation.schemas.<a href="./src/cloudflare/resources/schema_validation/schemas.py">get</a>(schema_id, \*, zone_id, \*\*<a href="src/cloudflare/types/schema_validation/schema_get_params.py">params</a>) -> <a href="./src/cloudflare/types/schema_validation/schema_get_response.py">SchemaGetResponse</a></code>
## Settings
Types:
```python
from cloudflare.types.schema_validation import (
SettingUpdateResponse,
SettingEditResponse,
SettingGetResponse,
)
```
Methods:
- <code title="put /zones/{zone_id}/schema_validation/settings">client.schema_validation.settings.<a href="./src/cloudflare/resources/schema_validation/settings/settings.py">update</a>(\*, zone_id, \*\*<a href="src/cloudflare/types/schema_validation/setting_update_params.py">params</a>) -> <a href="./src/cloudflare/types/schema_validation/setting_update_response.py">SettingUpdateResponse</a></code>
- <code title="patch /zones/{zone_id}/schema_validation/settings">client.schema_validation.settings.<a href="./src/cloudflare/resources/schema_validation/settings/settings.py">edit</a>(\*, zone_id, \*\*<a href="src/cloudflare/types/schema_validation/setting_edit_params.py">params</a>) -> <a href="./src/cloudflare/types/schema_validation/setting_edit_response.py">SettingEditResponse</a></code>
- <code title="get /zones/{zone_id}/schema_validation/settings">client.schema_validation.settings.<a href="./src/cloudflare/resources/schema_validation/settings/settings.py">get</a>(\*, zone_id) -> <a href="./src/cloudflare/types/schema_validation/setting_get_response.py">SettingGetResponse</a></code>
### Operations
Types:
```python
from cloudflare.types.schema_validation.settings import (
OperationUpdateResponse,
OperationListResponse,
OperationDeleteResponse,
OperationBulkEditResponse,
OperationGetResponse,
)
```
Methods:
- <code title="put /zones/{zone_id}/schema_validation/settings/operations/{operation_id}">client.schema_validation.settings.operations.<a href="./src/cloudflare/resources/schema_validation/settings/operations.py">update</a>(operation_id, \*, zone_id, \*\*<a href="src/cloudflare/types/schema_validation/settings/operation_update_params.py">params</a>) -> <a href="./src/cloudflare/types/schema_validation/settings/operation_update_response.py">OperationUpdateResponse</a></code>
- <code title="get /zones/{zone_id}/schema_validation/settings/operations">client.schema_validation.settings.operations.<a href="./src/cloudflare/resources/schema_validation/settings/operations.py">list</a>(\*, zone_id, \*\*<a href="src/cloudflare/types/schema_validation/settings/operation_list_params.py">params</a>) -> <a href="./src/cloudflare/types/schema_validation/settings/operation_list_response.py">SyncV4PagePaginationArray[OperationListResponse]</a></code>
- <code title="delete /zones/{zone_id}/schema_validation/settings/operations/{operation_id}">client.schema_validation.settings.operations.<a href="./src/cloudflare/resources/schema_validation/settings/operations.py">delete</a>(operation_id, \*, zone_id) -> <a href="./src/cloudflare/types/schema_validation/settings/operation_delete_response.py">OperationDeleteResponse</a></code>
- <code title="patch /zones/{zone_id}/schema_validation/settings/operations">client.schema_validation.settings.operations.<a href="./src/cloudflare/resources/schema_validation/settings/operations.py">bulk_edit</a>(\*, zone_id, \*\*<a href="src/cloudflare/types/schema_validation/settings/operation_bulk_edit_params.py">params</a>) -> <a href="./src/cloudflare/types/schema_validation/settings/operation_bulk_edit_response.py">OperationBulkEditResponse</a></code>
- <code title="get /zones/{zone_id}/schema_validation/settings/operations/{operation_id}">client.schema_validation.settings.operations.<a href="./src/cloudflare/resources/schema_validation/settings/operations.py">get</a>(operation_id, \*, zone_id) -> <a href="./src/cloudflare/types/schema_validation/settings/operation_get_response.py">OperationGetResponse</a></code>

View file

@ -113,6 +113,7 @@ if TYPE_CHECKING:
resource_sharing,
browser_rendering,
mtls_certificates,
schema_validation,
url_normalization,
custom_nameservers,
managed_transforms,
@ -218,6 +219,7 @@ if TYPE_CHECKING:
from .resources.resource_sharing.resource_sharing import ResourceSharingResource, AsyncResourceSharingResource
from .resources.browser_rendering.browser_rendering import BrowserRenderingResource, AsyncBrowserRenderingResource
from .resources.mtls_certificates.mtls_certificates import MTLSCertificatesResource, AsyncMTLSCertificatesResource
from .resources.schema_validation.schema_validation import SchemaValidationResource, AsyncSchemaValidationResource
from .resources.custom_certificates.custom_certificates import (
CustomCertificatesResource,
AsyncCustomCertificatesResource,
@ -905,6 +907,12 @@ class Cloudflare(SyncAPIClient):
return PipelinesResource(self)
@cached_property
def schema_validation(self) -> SchemaValidationResource:
from .resources.schema_validation import SchemaValidationResource
return SchemaValidationResource(self)
@cached_property
def with_raw_response(self) -> CloudflareWithRawResponse:
return CloudflareWithRawResponse(self)
@ -1731,6 +1739,12 @@ class AsyncCloudflare(AsyncAPIClient):
return AsyncPipelinesResource(self)
@cached_property
def schema_validation(self) -> AsyncSchemaValidationResource:
from .resources.schema_validation import AsyncSchemaValidationResource
return AsyncSchemaValidationResource(self)
@cached_property
def with_raw_response(self) -> AsyncCloudflareWithRawResponse:
return AsyncCloudflareWithRawResponse(self)
@ -2487,6 +2501,12 @@ class CloudflareWithRawResponse:
return PipelinesResourceWithRawResponse(self._client.pipelines)
@cached_property
def schema_validation(self) -> schema_validation.SchemaValidationResourceWithRawResponse:
from .resources.schema_validation import SchemaValidationResourceWithRawResponse
return SchemaValidationResourceWithRawResponse(self._client.schema_validation)
class AsyncCloudflareWithRawResponse:
_client: AsyncCloudflare
@ -3060,6 +3080,12 @@ class AsyncCloudflareWithRawResponse:
return AsyncPipelinesResourceWithRawResponse(self._client.pipelines)
@cached_property
def schema_validation(self) -> schema_validation.AsyncSchemaValidationResourceWithRawResponse:
from .resources.schema_validation import AsyncSchemaValidationResourceWithRawResponse
return AsyncSchemaValidationResourceWithRawResponse(self._client.schema_validation)
class CloudflareWithStreamedResponse:
_client: Cloudflare
@ -3633,6 +3659,12 @@ class CloudflareWithStreamedResponse:
return PipelinesResourceWithStreamingResponse(self._client.pipelines)
@cached_property
def schema_validation(self) -> schema_validation.SchemaValidationResourceWithStreamingResponse:
from .resources.schema_validation import SchemaValidationResourceWithStreamingResponse
return SchemaValidationResourceWithStreamingResponse(self._client.schema_validation)
class AsyncCloudflareWithStreamedResponse:
_client: AsyncCloudflare
@ -4216,6 +4248,12 @@ class AsyncCloudflareWithStreamedResponse:
return AsyncPipelinesResourceWithStreamingResponse(self._client.pipelines)
@cached_property
def schema_validation(self) -> schema_validation.AsyncSchemaValidationResourceWithStreamingResponse:
from .resources.schema_validation import AsyncSchemaValidationResourceWithStreamingResponse
return AsyncSchemaValidationResourceWithStreamingResponse(self._client.schema_validation)
Client = Cloudflare

View file

@ -0,0 +1,47 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from .schemas import (
SchemasResource,
AsyncSchemasResource,
SchemasResourceWithRawResponse,
AsyncSchemasResourceWithRawResponse,
SchemasResourceWithStreamingResponse,
AsyncSchemasResourceWithStreamingResponse,
)
from .settings import (
SettingsResource,
AsyncSettingsResource,
SettingsResourceWithRawResponse,
AsyncSettingsResourceWithRawResponse,
SettingsResourceWithStreamingResponse,
AsyncSettingsResourceWithStreamingResponse,
)
from .schema_validation import (
SchemaValidationResource,
AsyncSchemaValidationResource,
SchemaValidationResourceWithRawResponse,
AsyncSchemaValidationResourceWithRawResponse,
SchemaValidationResourceWithStreamingResponse,
AsyncSchemaValidationResourceWithStreamingResponse,
)
__all__ = [
"SchemasResource",
"AsyncSchemasResource",
"SchemasResourceWithRawResponse",
"AsyncSchemasResourceWithRawResponse",
"SchemasResourceWithStreamingResponse",
"AsyncSchemasResourceWithStreamingResponse",
"SettingsResource",
"AsyncSettingsResource",
"SettingsResourceWithRawResponse",
"AsyncSettingsResourceWithRawResponse",
"SettingsResourceWithStreamingResponse",
"AsyncSettingsResourceWithStreamingResponse",
"SchemaValidationResource",
"AsyncSchemaValidationResource",
"SchemaValidationResourceWithRawResponse",
"AsyncSchemaValidationResourceWithRawResponse",
"SchemaValidationResourceWithStreamingResponse",
"AsyncSchemaValidationResourceWithStreamingResponse",
]

View file

@ -0,0 +1,134 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from .schemas import (
SchemasResource,
AsyncSchemasResource,
SchemasResourceWithRawResponse,
AsyncSchemasResourceWithRawResponse,
SchemasResourceWithStreamingResponse,
AsyncSchemasResourceWithStreamingResponse,
)
from ..._compat import cached_property
from ..._resource import SyncAPIResource, AsyncAPIResource
from .settings.settings import (
SettingsResource,
AsyncSettingsResource,
SettingsResourceWithRawResponse,
AsyncSettingsResourceWithRawResponse,
SettingsResourceWithStreamingResponse,
AsyncSettingsResourceWithStreamingResponse,
)
__all__ = ["SchemaValidationResource", "AsyncSchemaValidationResource"]
class SchemaValidationResource(SyncAPIResource):
@cached_property
def schemas(self) -> SchemasResource:
return SchemasResource(self._client)
@cached_property
def settings(self) -> SettingsResource:
return SettingsResource(self._client)
@cached_property
def with_raw_response(self) -> SchemaValidationResourceWithRawResponse:
"""
This property can be used as a prefix for any HTTP method call to return
the raw response object instead of the parsed content.
For more information, see https://www.github.com/cloudflare/cloudflare-python#accessing-raw-response-data-eg-headers
"""
return SchemaValidationResourceWithRawResponse(self)
@cached_property
def with_streaming_response(self) -> SchemaValidationResourceWithStreamingResponse:
"""
An alternative to `.with_raw_response` that doesn't eagerly read the response body.
For more information, see https://www.github.com/cloudflare/cloudflare-python#with_streaming_response
"""
return SchemaValidationResourceWithStreamingResponse(self)
class AsyncSchemaValidationResource(AsyncAPIResource):
@cached_property
def schemas(self) -> AsyncSchemasResource:
return AsyncSchemasResource(self._client)
@cached_property
def settings(self) -> AsyncSettingsResource:
return AsyncSettingsResource(self._client)
@cached_property
def with_raw_response(self) -> AsyncSchemaValidationResourceWithRawResponse:
"""
This property can be used as a prefix for any HTTP method call to return
the raw response object instead of the parsed content.
For more information, see https://www.github.com/cloudflare/cloudflare-python#accessing-raw-response-data-eg-headers
"""
return AsyncSchemaValidationResourceWithRawResponse(self)
@cached_property
def with_streaming_response(self) -> AsyncSchemaValidationResourceWithStreamingResponse:
"""
An alternative to `.with_raw_response` that doesn't eagerly read the response body.
For more information, see https://www.github.com/cloudflare/cloudflare-python#with_streaming_response
"""
return AsyncSchemaValidationResourceWithStreamingResponse(self)
class SchemaValidationResourceWithRawResponse:
def __init__(self, schema_validation: SchemaValidationResource) -> None:
self._schema_validation = schema_validation
@cached_property
def schemas(self) -> SchemasResourceWithRawResponse:
return SchemasResourceWithRawResponse(self._schema_validation.schemas)
@cached_property
def settings(self) -> SettingsResourceWithRawResponse:
return SettingsResourceWithRawResponse(self._schema_validation.settings)
class AsyncSchemaValidationResourceWithRawResponse:
def __init__(self, schema_validation: AsyncSchemaValidationResource) -> None:
self._schema_validation = schema_validation
@cached_property
def schemas(self) -> AsyncSchemasResourceWithRawResponse:
return AsyncSchemasResourceWithRawResponse(self._schema_validation.schemas)
@cached_property
def settings(self) -> AsyncSettingsResourceWithRawResponse:
return AsyncSettingsResourceWithRawResponse(self._schema_validation.settings)
class SchemaValidationResourceWithStreamingResponse:
def __init__(self, schema_validation: SchemaValidationResource) -> None:
self._schema_validation = schema_validation
@cached_property
def schemas(self) -> SchemasResourceWithStreamingResponse:
return SchemasResourceWithStreamingResponse(self._schema_validation.schemas)
@cached_property
def settings(self) -> SettingsResourceWithStreamingResponse:
return SettingsResourceWithStreamingResponse(self._schema_validation.settings)
class AsyncSchemaValidationResourceWithStreamingResponse:
def __init__(self, schema_validation: AsyncSchemaValidationResource) -> None:
self._schema_validation = schema_validation
@cached_property
def schemas(self) -> AsyncSchemasResourceWithStreamingResponse:
return AsyncSchemasResourceWithStreamingResponse(self._schema_validation.schemas)
@cached_property
def settings(self) -> AsyncSettingsResourceWithStreamingResponse:
return AsyncSettingsResourceWithStreamingResponse(self._schema_validation.settings)

View file

@ -0,0 +1,678 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from typing import Type, cast
from typing_extensions import Literal
import httpx
from ..._types import NOT_GIVEN, Body, Query, Headers, NotGiven
from ..._utils import maybe_transform, async_maybe_transform
from ..._compat import cached_property
from ..._resource import SyncAPIResource, AsyncAPIResource
from ..._response import (
to_raw_response_wrapper,
to_streamed_response_wrapper,
async_to_raw_response_wrapper,
async_to_streamed_response_wrapper,
)
from ..._wrappers import ResultWrapper
from ...pagination import SyncV4PagePaginationArray, AsyncV4PagePaginationArray
from ..._base_client import AsyncPaginator, make_request_options
from ...types.schema_validation import schema_get_params, schema_edit_params, schema_list_params, schema_create_params
from ...types.schema_validation.schema_get_response import SchemaGetResponse
from ...types.schema_validation.schema_edit_response import SchemaEditResponse
from ...types.schema_validation.schema_list_response import SchemaListResponse
from ...types.schema_validation.schema_create_response import SchemaCreateResponse
from ...types.schema_validation.schema_delete_response import SchemaDeleteResponse
__all__ = ["SchemasResource", "AsyncSchemasResource"]
class SchemasResource(SyncAPIResource):
@cached_property
def with_raw_response(self) -> SchemasResourceWithRawResponse:
"""
This property can be used as a prefix for any HTTP method call to return
the raw response object instead of the parsed content.
For more information, see https://www.github.com/cloudflare/cloudflare-python#accessing-raw-response-data-eg-headers
"""
return SchemasResourceWithRawResponse(self)
@cached_property
def with_streaming_response(self) -> SchemasResourceWithStreamingResponse:
"""
An alternative to `.with_raw_response` that doesn't eagerly read the response body.
For more information, see https://www.github.com/cloudflare/cloudflare-python#with_streaming_response
"""
return SchemasResourceWithStreamingResponse(self)
def create(
self,
*,
zone_id: str,
kind: Literal["openapi_v3"],
name: str,
source: str,
validation_enabled: bool | NotGiven = NOT_GIVEN,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SchemaCreateResponse:
"""
Upload a schema
Args:
zone_id: Identifier.
kind: The kind of the schema
name: A human-readable name for the schema
source: The raw schema, e.g., the OpenAPI schema, either as JSON or YAML
validation_enabled: An indicator if this schema is enabled
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
return self._post(
f"/zones/{zone_id}/schema_validation/schemas",
body=maybe_transform(
{
"kind": kind,
"name": name,
"source": source,
"validation_enabled": validation_enabled,
},
schema_create_params.SchemaCreateParams,
),
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[SchemaCreateResponse]._unwrapper,
),
cast_to=cast(Type[SchemaCreateResponse], ResultWrapper[SchemaCreateResponse]),
)
def list(
self,
*,
zone_id: str,
omit_source: bool | NotGiven = NOT_GIVEN,
page: int | NotGiven = NOT_GIVEN,
per_page: int | NotGiven = NOT_GIVEN,
validation_enabled: bool | NotGiven = NOT_GIVEN,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SyncV4PagePaginationArray[SchemaListResponse]:
"""
List all uploaded schemas
Args:
zone_id: Identifier.
omit_source: Omit the source-files of schemas and only retrieve their meta-data.
page: Page number of paginated results.
per_page: Maximum number of results per page.
validation_enabled: Filter for enabled schemas
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
return self._get_api_list(
f"/zones/{zone_id}/schema_validation/schemas",
page=SyncV4PagePaginationArray[SchemaListResponse],
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
query=maybe_transform(
{
"omit_source": omit_source,
"page": page,
"per_page": per_page,
"validation_enabled": validation_enabled,
},
schema_list_params.SchemaListParams,
),
),
model=SchemaListResponse,
)
def delete(
self,
schema_id: str,
*,
zone_id: str,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SchemaDeleteResponse:
"""
Delete a schema
Args:
zone_id: Identifier.
schema_id: UUID.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
if not schema_id:
raise ValueError(f"Expected a non-empty value for `schema_id` but received {schema_id!r}")
return self._delete(
f"/zones/{zone_id}/schema_validation/schemas/{schema_id}",
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[SchemaDeleteResponse]._unwrapper,
),
cast_to=cast(Type[SchemaDeleteResponse], ResultWrapper[SchemaDeleteResponse]),
)
def edit(
self,
schema_id: str,
*,
zone_id: str,
validation_enabled: bool | NotGiven = NOT_GIVEN,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SchemaEditResponse:
"""
Edit details of a schema to enable validation
Args:
zone_id: Identifier.
schema_id: UUID.
validation_enabled: Flag whether schema is enabled for validation.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
if not schema_id:
raise ValueError(f"Expected a non-empty value for `schema_id` but received {schema_id!r}")
return self._patch(
f"/zones/{zone_id}/schema_validation/schemas/{schema_id}",
body=maybe_transform({"validation_enabled": validation_enabled}, schema_edit_params.SchemaEditParams),
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[SchemaEditResponse]._unwrapper,
),
cast_to=cast(Type[SchemaEditResponse], ResultWrapper[SchemaEditResponse]),
)
def get(
self,
schema_id: str,
*,
zone_id: str,
omit_source: bool | NotGiven = NOT_GIVEN,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SchemaGetResponse:
"""
Get details of a schema
Args:
zone_id: Identifier.
schema_id: UUID.
omit_source: Omit the source-files of schemas and only retrieve their meta-data.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
if not schema_id:
raise ValueError(f"Expected a non-empty value for `schema_id` but received {schema_id!r}")
return self._get(
f"/zones/{zone_id}/schema_validation/schemas/{schema_id}",
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
query=maybe_transform({"omit_source": omit_source}, schema_get_params.SchemaGetParams),
post_parser=ResultWrapper[SchemaGetResponse]._unwrapper,
),
cast_to=cast(Type[SchemaGetResponse], ResultWrapper[SchemaGetResponse]),
)
class AsyncSchemasResource(AsyncAPIResource):
@cached_property
def with_raw_response(self) -> AsyncSchemasResourceWithRawResponse:
"""
This property can be used as a prefix for any HTTP method call to return
the raw response object instead of the parsed content.
For more information, see https://www.github.com/cloudflare/cloudflare-python#accessing-raw-response-data-eg-headers
"""
return AsyncSchemasResourceWithRawResponse(self)
@cached_property
def with_streaming_response(self) -> AsyncSchemasResourceWithStreamingResponse:
"""
An alternative to `.with_raw_response` that doesn't eagerly read the response body.
For more information, see https://www.github.com/cloudflare/cloudflare-python#with_streaming_response
"""
return AsyncSchemasResourceWithStreamingResponse(self)
async def create(
self,
*,
zone_id: str,
kind: Literal["openapi_v3"],
name: str,
source: str,
validation_enabled: bool | NotGiven = NOT_GIVEN,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SchemaCreateResponse:
"""
Upload a schema
Args:
zone_id: Identifier.
kind: The kind of the schema
name: A human-readable name for the schema
source: The raw schema, e.g., the OpenAPI schema, either as JSON or YAML
validation_enabled: An indicator if this schema is enabled
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
return await self._post(
f"/zones/{zone_id}/schema_validation/schemas",
body=await async_maybe_transform(
{
"kind": kind,
"name": name,
"source": source,
"validation_enabled": validation_enabled,
},
schema_create_params.SchemaCreateParams,
),
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[SchemaCreateResponse]._unwrapper,
),
cast_to=cast(Type[SchemaCreateResponse], ResultWrapper[SchemaCreateResponse]),
)
def list(
self,
*,
zone_id: str,
omit_source: bool | NotGiven = NOT_GIVEN,
page: int | NotGiven = NOT_GIVEN,
per_page: int | NotGiven = NOT_GIVEN,
validation_enabled: bool | NotGiven = NOT_GIVEN,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> AsyncPaginator[SchemaListResponse, AsyncV4PagePaginationArray[SchemaListResponse]]:
"""
List all uploaded schemas
Args:
zone_id: Identifier.
omit_source: Omit the source-files of schemas and only retrieve their meta-data.
page: Page number of paginated results.
per_page: Maximum number of results per page.
validation_enabled: Filter for enabled schemas
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
return self._get_api_list(
f"/zones/{zone_id}/schema_validation/schemas",
page=AsyncV4PagePaginationArray[SchemaListResponse],
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
query=maybe_transform(
{
"omit_source": omit_source,
"page": page,
"per_page": per_page,
"validation_enabled": validation_enabled,
},
schema_list_params.SchemaListParams,
),
),
model=SchemaListResponse,
)
async def delete(
self,
schema_id: str,
*,
zone_id: str,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SchemaDeleteResponse:
"""
Delete a schema
Args:
zone_id: Identifier.
schema_id: UUID.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
if not schema_id:
raise ValueError(f"Expected a non-empty value for `schema_id` but received {schema_id!r}")
return await self._delete(
f"/zones/{zone_id}/schema_validation/schemas/{schema_id}",
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[SchemaDeleteResponse]._unwrapper,
),
cast_to=cast(Type[SchemaDeleteResponse], ResultWrapper[SchemaDeleteResponse]),
)
async def edit(
self,
schema_id: str,
*,
zone_id: str,
validation_enabled: bool | NotGiven = NOT_GIVEN,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SchemaEditResponse:
"""
Edit details of a schema to enable validation
Args:
zone_id: Identifier.
schema_id: UUID.
validation_enabled: Flag whether schema is enabled for validation.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
if not schema_id:
raise ValueError(f"Expected a non-empty value for `schema_id` but received {schema_id!r}")
return await self._patch(
f"/zones/{zone_id}/schema_validation/schemas/{schema_id}",
body=await async_maybe_transform(
{"validation_enabled": validation_enabled}, schema_edit_params.SchemaEditParams
),
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[SchemaEditResponse]._unwrapper,
),
cast_to=cast(Type[SchemaEditResponse], ResultWrapper[SchemaEditResponse]),
)
async def get(
self,
schema_id: str,
*,
zone_id: str,
omit_source: bool | NotGiven = NOT_GIVEN,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SchemaGetResponse:
"""
Get details of a schema
Args:
zone_id: Identifier.
schema_id: UUID.
omit_source: Omit the source-files of schemas and only retrieve their meta-data.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
if not schema_id:
raise ValueError(f"Expected a non-empty value for `schema_id` but received {schema_id!r}")
return await self._get(
f"/zones/{zone_id}/schema_validation/schemas/{schema_id}",
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
query=await async_maybe_transform({"omit_source": omit_source}, schema_get_params.SchemaGetParams),
post_parser=ResultWrapper[SchemaGetResponse]._unwrapper,
),
cast_to=cast(Type[SchemaGetResponse], ResultWrapper[SchemaGetResponse]),
)
class SchemasResourceWithRawResponse:
def __init__(self, schemas: SchemasResource) -> None:
self._schemas = schemas
self.create = to_raw_response_wrapper(
schemas.create,
)
self.list = to_raw_response_wrapper(
schemas.list,
)
self.delete = to_raw_response_wrapper(
schemas.delete,
)
self.edit = to_raw_response_wrapper(
schemas.edit,
)
self.get = to_raw_response_wrapper(
schemas.get,
)
class AsyncSchemasResourceWithRawResponse:
def __init__(self, schemas: AsyncSchemasResource) -> None:
self._schemas = schemas
self.create = async_to_raw_response_wrapper(
schemas.create,
)
self.list = async_to_raw_response_wrapper(
schemas.list,
)
self.delete = async_to_raw_response_wrapper(
schemas.delete,
)
self.edit = async_to_raw_response_wrapper(
schemas.edit,
)
self.get = async_to_raw_response_wrapper(
schemas.get,
)
class SchemasResourceWithStreamingResponse:
def __init__(self, schemas: SchemasResource) -> None:
self._schemas = schemas
self.create = to_streamed_response_wrapper(
schemas.create,
)
self.list = to_streamed_response_wrapper(
schemas.list,
)
self.delete = to_streamed_response_wrapper(
schemas.delete,
)
self.edit = to_streamed_response_wrapper(
schemas.edit,
)
self.get = to_streamed_response_wrapper(
schemas.get,
)
class AsyncSchemasResourceWithStreamingResponse:
def __init__(self, schemas: AsyncSchemasResource) -> None:
self._schemas = schemas
self.create = async_to_streamed_response_wrapper(
schemas.create,
)
self.list = async_to_streamed_response_wrapper(
schemas.list,
)
self.delete = async_to_streamed_response_wrapper(
schemas.delete,
)
self.edit = async_to_streamed_response_wrapper(
schemas.edit,
)
self.get = async_to_streamed_response_wrapper(
schemas.get,
)

View file

@ -0,0 +1,33 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from .settings import (
SettingsResource,
AsyncSettingsResource,
SettingsResourceWithRawResponse,
AsyncSettingsResourceWithRawResponse,
SettingsResourceWithStreamingResponse,
AsyncSettingsResourceWithStreamingResponse,
)
from .operations import (
OperationsResource,
AsyncOperationsResource,
OperationsResourceWithRawResponse,
AsyncOperationsResourceWithRawResponse,
OperationsResourceWithStreamingResponse,
AsyncOperationsResourceWithStreamingResponse,
)
__all__ = [
"OperationsResource",
"AsyncOperationsResource",
"OperationsResourceWithRawResponse",
"AsyncOperationsResourceWithRawResponse",
"OperationsResourceWithStreamingResponse",
"AsyncOperationsResourceWithStreamingResponse",
"SettingsResource",
"AsyncSettingsResource",
"SettingsResourceWithRawResponse",
"AsyncSettingsResourceWithRawResponse",
"SettingsResourceWithStreamingResponse",
"AsyncSettingsResourceWithStreamingResponse",
]

View file

@ -0,0 +1,636 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from typing import Dict, Type, Optional, cast
from typing_extensions import Literal
import httpx
from ...._types import NOT_GIVEN, Body, Query, Headers, NotGiven
from ...._utils import maybe_transform, async_maybe_transform
from ...._compat import cached_property
from ...._resource import SyncAPIResource, AsyncAPIResource
from ...._response import (
to_raw_response_wrapper,
to_streamed_response_wrapper,
async_to_raw_response_wrapper,
async_to_streamed_response_wrapper,
)
from ...._wrappers import ResultWrapper
from ....pagination import SyncV4PagePaginationArray, AsyncV4PagePaginationArray
from ...._base_client import AsyncPaginator, make_request_options
from ....types.schema_validation.settings import (
operation_list_params,
operation_update_params,
operation_bulk_edit_params,
)
from ....types.schema_validation.settings.operation_get_response import OperationGetResponse
from ....types.schema_validation.settings.operation_list_response import OperationListResponse
from ....types.schema_validation.settings.operation_delete_response import OperationDeleteResponse
from ....types.schema_validation.settings.operation_update_response import OperationUpdateResponse
from ....types.schema_validation.settings.operation_bulk_edit_response import OperationBulkEditResponse
__all__ = ["OperationsResource", "AsyncOperationsResource"]
class OperationsResource(SyncAPIResource):
@cached_property
def with_raw_response(self) -> OperationsResourceWithRawResponse:
"""
This property can be used as a prefix for any HTTP method call to return
the raw response object instead of the parsed content.
For more information, see https://www.github.com/cloudflare/cloudflare-python#accessing-raw-response-data-eg-headers
"""
return OperationsResourceWithRawResponse(self)
@cached_property
def with_streaming_response(self) -> OperationsResourceWithStreamingResponse:
"""
An alternative to `.with_raw_response` that doesn't eagerly read the response body.
For more information, see https://www.github.com/cloudflare/cloudflare-python#with_streaming_response
"""
return OperationsResourceWithStreamingResponse(self)
def update(
self,
operation_id: str,
*,
zone_id: str,
mitigation_action: Optional[Literal["log", "block", "none"]],
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> OperationUpdateResponse:
"""
Update per-operation schema validation setting
Args:
zone_id: Identifier.
operation_id: UUID.
mitigation_action: When set, this applies a mitigation action to this operation
- `"log"` - log request when request does not conform to schema for this
operation
- `"block"` - deny access to the site when request does not conform to schema
for this operation
- `"none"` - will skip mitigation for this operation
- `null` - clears any mitigation action
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
if not operation_id:
raise ValueError(f"Expected a non-empty value for `operation_id` but received {operation_id!r}")
return self._put(
f"/zones/{zone_id}/schema_validation/settings/operations/{operation_id}",
body=maybe_transform(
{"mitigation_action": mitigation_action}, operation_update_params.OperationUpdateParams
),
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[OperationUpdateResponse]._unwrapper,
),
cast_to=cast(Type[OperationUpdateResponse], ResultWrapper[OperationUpdateResponse]),
)
def list(
self,
*,
zone_id: str,
page: int | NotGiven = NOT_GIVEN,
per_page: int | NotGiven = NOT_GIVEN,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SyncV4PagePaginationArray[OperationListResponse]:
"""
List per-operation schema validation settings
Args:
zone_id: Identifier.
page: Page number of paginated results.
per_page: Maximum number of results per page.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
return self._get_api_list(
f"/zones/{zone_id}/schema_validation/settings/operations",
page=SyncV4PagePaginationArray[OperationListResponse],
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
query=maybe_transform(
{
"page": page,
"per_page": per_page,
},
operation_list_params.OperationListParams,
),
),
model=OperationListResponse,
)
def delete(
self,
operation_id: str,
*,
zone_id: str,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> OperationDeleteResponse:
"""
Delete per-operation schema validation setting
Args:
zone_id: Identifier.
operation_id: UUID.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
if not operation_id:
raise ValueError(f"Expected a non-empty value for `operation_id` but received {operation_id!r}")
return self._delete(
f"/zones/{zone_id}/schema_validation/settings/operations/{operation_id}",
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[OperationDeleteResponse]._unwrapper,
),
cast_to=cast(Type[OperationDeleteResponse], ResultWrapper[OperationDeleteResponse]),
)
def bulk_edit(
self,
*,
zone_id: str,
body: Dict[str, operation_bulk_edit_params.Body],
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> OperationBulkEditResponse:
"""
Bulk edit per-operation schema validation settings
Args:
zone_id: Identifier.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
return self._patch(
f"/zones/{zone_id}/schema_validation/settings/operations",
body=maybe_transform(body, operation_bulk_edit_params.OperationBulkEditParams),
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[OperationBulkEditResponse]._unwrapper,
),
cast_to=cast(Type[OperationBulkEditResponse], ResultWrapper[OperationBulkEditResponse]),
)
def get(
self,
operation_id: str,
*,
zone_id: str,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> OperationGetResponse:
"""
Get per-operation schema validation setting
Args:
zone_id: Identifier.
operation_id: UUID.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
if not operation_id:
raise ValueError(f"Expected a non-empty value for `operation_id` but received {operation_id!r}")
return self._get(
f"/zones/{zone_id}/schema_validation/settings/operations/{operation_id}",
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[OperationGetResponse]._unwrapper,
),
cast_to=cast(Type[OperationGetResponse], ResultWrapper[OperationGetResponse]),
)
class AsyncOperationsResource(AsyncAPIResource):
@cached_property
def with_raw_response(self) -> AsyncOperationsResourceWithRawResponse:
"""
This property can be used as a prefix for any HTTP method call to return
the raw response object instead of the parsed content.
For more information, see https://www.github.com/cloudflare/cloudflare-python#accessing-raw-response-data-eg-headers
"""
return AsyncOperationsResourceWithRawResponse(self)
@cached_property
def with_streaming_response(self) -> AsyncOperationsResourceWithStreamingResponse:
"""
An alternative to `.with_raw_response` that doesn't eagerly read the response body.
For more information, see https://www.github.com/cloudflare/cloudflare-python#with_streaming_response
"""
return AsyncOperationsResourceWithStreamingResponse(self)
async def update(
self,
operation_id: str,
*,
zone_id: str,
mitigation_action: Optional[Literal["log", "block", "none"]],
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> OperationUpdateResponse:
"""
Update per-operation schema validation setting
Args:
zone_id: Identifier.
operation_id: UUID.
mitigation_action: When set, this applies a mitigation action to this operation
- `"log"` - log request when request does not conform to schema for this
operation
- `"block"` - deny access to the site when request does not conform to schema
for this operation
- `"none"` - will skip mitigation for this operation
- `null` - clears any mitigation action
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
if not operation_id:
raise ValueError(f"Expected a non-empty value for `operation_id` but received {operation_id!r}")
return await self._put(
f"/zones/{zone_id}/schema_validation/settings/operations/{operation_id}",
body=await async_maybe_transform(
{"mitigation_action": mitigation_action}, operation_update_params.OperationUpdateParams
),
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[OperationUpdateResponse]._unwrapper,
),
cast_to=cast(Type[OperationUpdateResponse], ResultWrapper[OperationUpdateResponse]),
)
def list(
self,
*,
zone_id: str,
page: int | NotGiven = NOT_GIVEN,
per_page: int | NotGiven = NOT_GIVEN,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> AsyncPaginator[OperationListResponse, AsyncV4PagePaginationArray[OperationListResponse]]:
"""
List per-operation schema validation settings
Args:
zone_id: Identifier.
page: Page number of paginated results.
per_page: Maximum number of results per page.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
return self._get_api_list(
f"/zones/{zone_id}/schema_validation/settings/operations",
page=AsyncV4PagePaginationArray[OperationListResponse],
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
query=maybe_transform(
{
"page": page,
"per_page": per_page,
},
operation_list_params.OperationListParams,
),
),
model=OperationListResponse,
)
async def delete(
self,
operation_id: str,
*,
zone_id: str,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> OperationDeleteResponse:
"""
Delete per-operation schema validation setting
Args:
zone_id: Identifier.
operation_id: UUID.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
if not operation_id:
raise ValueError(f"Expected a non-empty value for `operation_id` but received {operation_id!r}")
return await self._delete(
f"/zones/{zone_id}/schema_validation/settings/operations/{operation_id}",
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[OperationDeleteResponse]._unwrapper,
),
cast_to=cast(Type[OperationDeleteResponse], ResultWrapper[OperationDeleteResponse]),
)
async def bulk_edit(
self,
*,
zone_id: str,
body: Dict[str, operation_bulk_edit_params.Body],
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> OperationBulkEditResponse:
"""
Bulk edit per-operation schema validation settings
Args:
zone_id: Identifier.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
return await self._patch(
f"/zones/{zone_id}/schema_validation/settings/operations",
body=await async_maybe_transform(body, operation_bulk_edit_params.OperationBulkEditParams),
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[OperationBulkEditResponse]._unwrapper,
),
cast_to=cast(Type[OperationBulkEditResponse], ResultWrapper[OperationBulkEditResponse]),
)
async def get(
self,
operation_id: str,
*,
zone_id: str,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> OperationGetResponse:
"""
Get per-operation schema validation setting
Args:
zone_id: Identifier.
operation_id: UUID.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
if not operation_id:
raise ValueError(f"Expected a non-empty value for `operation_id` but received {operation_id!r}")
return await self._get(
f"/zones/{zone_id}/schema_validation/settings/operations/{operation_id}",
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[OperationGetResponse]._unwrapper,
),
cast_to=cast(Type[OperationGetResponse], ResultWrapper[OperationGetResponse]),
)
class OperationsResourceWithRawResponse:
def __init__(self, operations: OperationsResource) -> None:
self._operations = operations
self.update = to_raw_response_wrapper(
operations.update,
)
self.list = to_raw_response_wrapper(
operations.list,
)
self.delete = to_raw_response_wrapper(
operations.delete,
)
self.bulk_edit = to_raw_response_wrapper(
operations.bulk_edit,
)
self.get = to_raw_response_wrapper(
operations.get,
)
class AsyncOperationsResourceWithRawResponse:
def __init__(self, operations: AsyncOperationsResource) -> None:
self._operations = operations
self.update = async_to_raw_response_wrapper(
operations.update,
)
self.list = async_to_raw_response_wrapper(
operations.list,
)
self.delete = async_to_raw_response_wrapper(
operations.delete,
)
self.bulk_edit = async_to_raw_response_wrapper(
operations.bulk_edit,
)
self.get = async_to_raw_response_wrapper(
operations.get,
)
class OperationsResourceWithStreamingResponse:
def __init__(self, operations: OperationsResource) -> None:
self._operations = operations
self.update = to_streamed_response_wrapper(
operations.update,
)
self.list = to_streamed_response_wrapper(
operations.list,
)
self.delete = to_streamed_response_wrapper(
operations.delete,
)
self.bulk_edit = to_streamed_response_wrapper(
operations.bulk_edit,
)
self.get = to_streamed_response_wrapper(
operations.get,
)
class AsyncOperationsResourceWithStreamingResponse:
def __init__(self, operations: AsyncOperationsResource) -> None:
self._operations = operations
self.update = async_to_streamed_response_wrapper(
operations.update,
)
self.list = async_to_streamed_response_wrapper(
operations.list,
)
self.delete = async_to_streamed_response_wrapper(
operations.delete,
)
self.bulk_edit = async_to_streamed_response_wrapper(
operations.bulk_edit,
)
self.get = async_to_streamed_response_wrapper(
operations.get,
)

View file

@ -0,0 +1,479 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from typing import Type, Optional, cast
from typing_extensions import Literal
import httpx
from ...._types import NOT_GIVEN, Body, Query, Headers, NotGiven
from ...._utils import maybe_transform, async_maybe_transform
from ...._compat import cached_property
from .operations import (
OperationsResource,
AsyncOperationsResource,
OperationsResourceWithRawResponse,
AsyncOperationsResourceWithRawResponse,
OperationsResourceWithStreamingResponse,
AsyncOperationsResourceWithStreamingResponse,
)
from ...._resource import SyncAPIResource, AsyncAPIResource
from ...._response import (
to_raw_response_wrapper,
to_streamed_response_wrapper,
async_to_raw_response_wrapper,
async_to_streamed_response_wrapper,
)
from ...._wrappers import ResultWrapper
from ...._base_client import make_request_options
from ....types.schema_validation import setting_edit_params, setting_update_params
from ....types.schema_validation.setting_get_response import SettingGetResponse
from ....types.schema_validation.setting_edit_response import SettingEditResponse
from ....types.schema_validation.setting_update_response import SettingUpdateResponse
__all__ = ["SettingsResource", "AsyncSettingsResource"]
class SettingsResource(SyncAPIResource):
@cached_property
def operations(self) -> OperationsResource:
return OperationsResource(self._client)
@cached_property
def with_raw_response(self) -> SettingsResourceWithRawResponse:
"""
This property can be used as a prefix for any HTTP method call to return
the raw response object instead of the parsed content.
For more information, see https://www.github.com/cloudflare/cloudflare-python#accessing-raw-response-data-eg-headers
"""
return SettingsResourceWithRawResponse(self)
@cached_property
def with_streaming_response(self) -> SettingsResourceWithStreamingResponse:
"""
An alternative to `.with_raw_response` that doesn't eagerly read the response body.
For more information, see https://www.github.com/cloudflare/cloudflare-python#with_streaming_response
"""
return SettingsResourceWithStreamingResponse(self)
def update(
self,
*,
zone_id: str,
validation_default_mitigation_action: Literal["none", "log", "block"],
validation_override_mitigation_action: Optional[Literal["none"]] | NotGiven = NOT_GIVEN,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SettingUpdateResponse:
"""
Update global schema validation settings
Args:
zone_id: Identifier.
validation_default_mitigation_action:
The default mitigation action used Mitigation actions are as follows:
- `"log"` - log request when request does not conform to schema
- `"block"` - deny access to the site when request does not conform to schema
- `"none"` - skip running schema validation
validation_override_mitigation_action: When set, this overrides both zone level and operation level mitigation actions.
- `"none"` - skip running schema validation entirely for the request
- `null` - clears any existing override
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
return self._put(
f"/zones/{zone_id}/schema_validation/settings",
body=maybe_transform(
{
"validation_default_mitigation_action": validation_default_mitigation_action,
"validation_override_mitigation_action": validation_override_mitigation_action,
},
setting_update_params.SettingUpdateParams,
),
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[SettingUpdateResponse]._unwrapper,
),
cast_to=cast(Type[SettingUpdateResponse], ResultWrapper[SettingUpdateResponse]),
)
def edit(
self,
*,
zone_id: str,
validation_default_mitigation_action: Literal["none", "log", "block"] | NotGiven = NOT_GIVEN,
validation_override_mitigation_action: Optional[Literal["none"]] | NotGiven = NOT_GIVEN,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SettingEditResponse:
"""
Edit global schema validation settings
Args:
zone_id: Identifier.
validation_default_mitigation_action:
The default mitigation action used Mitigation actions are as follows:
- `"log"` - log request when request does not conform to schema
- `"block"` - deny access to the site when request does not conform to schema
- `"none"` - skip running schema validation
validation_override_mitigation_action: When set, this overrides both zone level and operation level mitigation actions.
- `"none"` - skip running schema validation entirely for the request
- `null` - clears any existing override
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
return self._patch(
f"/zones/{zone_id}/schema_validation/settings",
body=maybe_transform(
{
"validation_default_mitigation_action": validation_default_mitigation_action,
"validation_override_mitigation_action": validation_override_mitigation_action,
},
setting_edit_params.SettingEditParams,
),
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[SettingEditResponse]._unwrapper,
),
cast_to=cast(Type[SettingEditResponse], ResultWrapper[SettingEditResponse]),
)
def get(
self,
*,
zone_id: str,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SettingGetResponse:
"""
Get global schema validation settings
Args:
zone_id: Identifier.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
return self._get(
f"/zones/{zone_id}/schema_validation/settings",
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[SettingGetResponse]._unwrapper,
),
cast_to=cast(Type[SettingGetResponse], ResultWrapper[SettingGetResponse]),
)
class AsyncSettingsResource(AsyncAPIResource):
@cached_property
def operations(self) -> AsyncOperationsResource:
return AsyncOperationsResource(self._client)
@cached_property
def with_raw_response(self) -> AsyncSettingsResourceWithRawResponse:
"""
This property can be used as a prefix for any HTTP method call to return
the raw response object instead of the parsed content.
For more information, see https://www.github.com/cloudflare/cloudflare-python#accessing-raw-response-data-eg-headers
"""
return AsyncSettingsResourceWithRawResponse(self)
@cached_property
def with_streaming_response(self) -> AsyncSettingsResourceWithStreamingResponse:
"""
An alternative to `.with_raw_response` that doesn't eagerly read the response body.
For more information, see https://www.github.com/cloudflare/cloudflare-python#with_streaming_response
"""
return AsyncSettingsResourceWithStreamingResponse(self)
async def update(
self,
*,
zone_id: str,
validation_default_mitigation_action: Literal["none", "log", "block"],
validation_override_mitigation_action: Optional[Literal["none"]] | NotGiven = NOT_GIVEN,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SettingUpdateResponse:
"""
Update global schema validation settings
Args:
zone_id: Identifier.
validation_default_mitigation_action:
The default mitigation action used Mitigation actions are as follows:
- `"log"` - log request when request does not conform to schema
- `"block"` - deny access to the site when request does not conform to schema
- `"none"` - skip running schema validation
validation_override_mitigation_action: When set, this overrides both zone level and operation level mitigation actions.
- `"none"` - skip running schema validation entirely for the request
- `null` - clears any existing override
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
return await self._put(
f"/zones/{zone_id}/schema_validation/settings",
body=await async_maybe_transform(
{
"validation_default_mitigation_action": validation_default_mitigation_action,
"validation_override_mitigation_action": validation_override_mitigation_action,
},
setting_update_params.SettingUpdateParams,
),
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[SettingUpdateResponse]._unwrapper,
),
cast_to=cast(Type[SettingUpdateResponse], ResultWrapper[SettingUpdateResponse]),
)
async def edit(
self,
*,
zone_id: str,
validation_default_mitigation_action: Literal["none", "log", "block"] | NotGiven = NOT_GIVEN,
validation_override_mitigation_action: Optional[Literal["none"]] | NotGiven = NOT_GIVEN,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SettingEditResponse:
"""
Edit global schema validation settings
Args:
zone_id: Identifier.
validation_default_mitigation_action:
The default mitigation action used Mitigation actions are as follows:
- `"log"` - log request when request does not conform to schema
- `"block"` - deny access to the site when request does not conform to schema
- `"none"` - skip running schema validation
validation_override_mitigation_action: When set, this overrides both zone level and operation level mitigation actions.
- `"none"` - skip running schema validation entirely for the request
- `null` - clears any existing override
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
return await self._patch(
f"/zones/{zone_id}/schema_validation/settings",
body=await async_maybe_transform(
{
"validation_default_mitigation_action": validation_default_mitigation_action,
"validation_override_mitigation_action": validation_override_mitigation_action,
},
setting_edit_params.SettingEditParams,
),
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[SettingEditResponse]._unwrapper,
),
cast_to=cast(Type[SettingEditResponse], ResultWrapper[SettingEditResponse]),
)
async def get(
self,
*,
zone_id: str,
# Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs.
# The extra values given here take precedence over values defined on the client or passed to this method.
extra_headers: Headers | None = None,
extra_query: Query | None = None,
extra_body: Body | None = None,
timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
) -> SettingGetResponse:
"""
Get global schema validation settings
Args:
zone_id: Identifier.
extra_headers: Send extra headers
extra_query: Add additional query parameters to the request
extra_body: Add additional JSON properties to the request
timeout: Override the client-level default timeout for this request, in seconds
"""
if not zone_id:
raise ValueError(f"Expected a non-empty value for `zone_id` but received {zone_id!r}")
return await self._get(
f"/zones/{zone_id}/schema_validation/settings",
options=make_request_options(
extra_headers=extra_headers,
extra_query=extra_query,
extra_body=extra_body,
timeout=timeout,
post_parser=ResultWrapper[SettingGetResponse]._unwrapper,
),
cast_to=cast(Type[SettingGetResponse], ResultWrapper[SettingGetResponse]),
)
class SettingsResourceWithRawResponse:
def __init__(self, settings: SettingsResource) -> None:
self._settings = settings
self.update = to_raw_response_wrapper(
settings.update,
)
self.edit = to_raw_response_wrapper(
settings.edit,
)
self.get = to_raw_response_wrapper(
settings.get,
)
@cached_property
def operations(self) -> OperationsResourceWithRawResponse:
return OperationsResourceWithRawResponse(self._settings.operations)
class AsyncSettingsResourceWithRawResponse:
def __init__(self, settings: AsyncSettingsResource) -> None:
self._settings = settings
self.update = async_to_raw_response_wrapper(
settings.update,
)
self.edit = async_to_raw_response_wrapper(
settings.edit,
)
self.get = async_to_raw_response_wrapper(
settings.get,
)
@cached_property
def operations(self) -> AsyncOperationsResourceWithRawResponse:
return AsyncOperationsResourceWithRawResponse(self._settings.operations)
class SettingsResourceWithStreamingResponse:
def __init__(self, settings: SettingsResource) -> None:
self._settings = settings
self.update = to_streamed_response_wrapper(
settings.update,
)
self.edit = to_streamed_response_wrapper(
settings.edit,
)
self.get = to_streamed_response_wrapper(
settings.get,
)
@cached_property
def operations(self) -> OperationsResourceWithStreamingResponse:
return OperationsResourceWithStreamingResponse(self._settings.operations)
class AsyncSettingsResourceWithStreamingResponse:
def __init__(self, settings: AsyncSettingsResource) -> None:
self._settings = settings
self.update = async_to_streamed_response_wrapper(
settings.update,
)
self.edit = async_to_streamed_response_wrapper(
settings.edit,
)
self.get = async_to_streamed_response_wrapper(
settings.get,
)
@cached_property
def operations(self) -> AsyncOperationsResourceWithStreamingResponse:
return AsyncOperationsResourceWithStreamingResponse(self._settings.operations)

View file

@ -0,0 +1,18 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from .schema_get_params import SchemaGetParams as SchemaGetParams
from .schema_edit_params import SchemaEditParams as SchemaEditParams
from .schema_list_params import SchemaListParams as SchemaListParams
from .schema_get_response import SchemaGetResponse as SchemaGetResponse
from .setting_edit_params import SettingEditParams as SettingEditParams
from .schema_create_params import SchemaCreateParams as SchemaCreateParams
from .schema_edit_response import SchemaEditResponse as SchemaEditResponse
from .schema_list_response import SchemaListResponse as SchemaListResponse
from .setting_get_response import SettingGetResponse as SettingGetResponse
from .setting_edit_response import SettingEditResponse as SettingEditResponse
from .setting_update_params import SettingUpdateParams as SettingUpdateParams
from .schema_create_response import SchemaCreateResponse as SchemaCreateResponse
from .schema_delete_response import SchemaDeleteResponse as SchemaDeleteResponse
from .setting_update_response import SettingUpdateResponse as SettingUpdateResponse

View file

@ -0,0 +1,24 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from typing_extensions import Literal, Required, TypedDict
__all__ = ["SchemaCreateParams"]
class SchemaCreateParams(TypedDict, total=False):
zone_id: Required[str]
"""Identifier."""
kind: Required[Literal["openapi_v3"]]
"""The kind of the schema"""
name: Required[str]
"""A human-readable name for the schema"""
source: Required[str]
"""The raw schema, e.g., the OpenAPI schema, either as JSON or YAML"""
validation_enabled: bool
"""An indicator if this schema is enabled"""

View file

@ -0,0 +1,28 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from typing import Optional
from datetime import datetime
from typing_extensions import Literal
from ..._models import BaseModel
__all__ = ["SchemaCreateResponse"]
class SchemaCreateResponse(BaseModel):
created_at: datetime
kind: Literal["openapi_v3"]
"""The kind of the schema"""
name: str
"""A human-readable name for the schema"""
schema_id: str
"""A unique identifier of this schema"""
source: str
"""The raw schema, e.g., the OpenAPI schema, either as JSON or YAML"""
validation_enabled: Optional[bool] = None
"""An indicator if this schema is enabled"""

View file

@ -0,0 +1,10 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from ..._models import BaseModel
__all__ = ["SchemaDeleteResponse"]
class SchemaDeleteResponse(BaseModel):
schema_id: str
"""The ID of the schema that was just deleted"""

View file

@ -0,0 +1,15 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from typing_extensions import Required, TypedDict
__all__ = ["SchemaEditParams"]
class SchemaEditParams(TypedDict, total=False):
zone_id: Required[str]
"""Identifier."""
validation_enabled: bool
"""Flag whether schema is enabled for validation."""

View file

@ -0,0 +1,28 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from typing import Optional
from datetime import datetime
from typing_extensions import Literal
from ..._models import BaseModel
__all__ = ["SchemaEditResponse"]
class SchemaEditResponse(BaseModel):
created_at: datetime
kind: Literal["openapi_v3"]
"""The kind of the schema"""
name: str
"""A human-readable name for the schema"""
schema_id: str
"""A unique identifier of this schema"""
source: str
"""The raw schema, e.g., the OpenAPI schema, either as JSON or YAML"""
validation_enabled: Optional[bool] = None
"""An indicator if this schema is enabled"""

View file

@ -0,0 +1,15 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from typing_extensions import Required, TypedDict
__all__ = ["SchemaGetParams"]
class SchemaGetParams(TypedDict, total=False):
zone_id: Required[str]
"""Identifier."""
omit_source: bool
"""Omit the source-files of schemas and only retrieve their meta-data."""

View file

@ -0,0 +1,28 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from typing import Optional
from datetime import datetime
from typing_extensions import Literal
from ..._models import BaseModel
__all__ = ["SchemaGetResponse"]
class SchemaGetResponse(BaseModel):
created_at: datetime
kind: Literal["openapi_v3"]
"""The kind of the schema"""
name: str
"""A human-readable name for the schema"""
schema_id: str
"""A unique identifier of this schema"""
source: str
"""The raw schema, e.g., the OpenAPI schema, either as JSON or YAML"""
validation_enabled: Optional[bool] = None
"""An indicator if this schema is enabled"""

View file

@ -0,0 +1,24 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from typing_extensions import Required, TypedDict
__all__ = ["SchemaListParams"]
class SchemaListParams(TypedDict, total=False):
zone_id: Required[str]
"""Identifier."""
omit_source: bool
"""Omit the source-files of schemas and only retrieve their meta-data."""
page: int
"""Page number of paginated results."""
per_page: int
"""Maximum number of results per page."""
validation_enabled: bool
"""Filter for enabled schemas"""

View file

@ -0,0 +1,28 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from typing import Optional
from datetime import datetime
from typing_extensions import Literal
from ..._models import BaseModel
__all__ = ["SchemaListResponse"]
class SchemaListResponse(BaseModel):
created_at: datetime
kind: Literal["openapi_v3"]
"""The kind of the schema"""
name: str
"""A human-readable name for the schema"""
schema_id: str
"""A unique identifier of this schema"""
source: str
"""The raw schema, e.g., the OpenAPI schema, either as JSON or YAML"""
validation_enabled: Optional[bool] = None
"""An indicator if this schema is enabled"""

View file

@ -0,0 +1,28 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from typing import Optional
from typing_extensions import Literal, Required, TypedDict
__all__ = ["SettingEditParams"]
class SettingEditParams(TypedDict, total=False):
zone_id: Required[str]
"""Identifier."""
validation_default_mitigation_action: Literal["none", "log", "block"]
"""The default mitigation action used Mitigation actions are as follows:
- `"log"` - log request when request does not conform to schema
- `"block"` - deny access to the site when request does not conform to schema
- `"none"` - skip running schema validation
"""
validation_override_mitigation_action: Optional[Literal["none"]]
"""When set, this overrides both zone level and operation level mitigation actions.
- `"none"` - skip running schema validation entirely for the request
- `null` - clears any existing override
"""

View file

@ -0,0 +1,29 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from typing import Optional
from typing_extensions import Literal
from ..._models import BaseModel
__all__ = ["SettingEditResponse"]
class SettingEditResponse(BaseModel):
validation_default_mitigation_action: Literal["none", "log", "block"]
"""The default mitigation action used
Mitigation actions are as follows:
- `log` - log request when request does not conform to schema
- `block` - deny access to the site when request does not conform to schema
- `none` - skip running schema validation
"""
validation_override_mitigation_action: Optional[Literal["none"]] = None
"""
When not null, this overrides global both zone level and operation level
mitigation actions. This can serve as a quick way to disable schema validation
for the whole zone.
- `"none"` will skip running schema validation entirely for the request
"""

View file

@ -0,0 +1,29 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from typing import Optional
from typing_extensions import Literal
from ..._models import BaseModel
__all__ = ["SettingGetResponse"]
class SettingGetResponse(BaseModel):
validation_default_mitigation_action: Literal["none", "log", "block"]
"""The default mitigation action used
Mitigation actions are as follows:
- `log` - log request when request does not conform to schema
- `block` - deny access to the site when request does not conform to schema
- `none` - skip running schema validation
"""
validation_override_mitigation_action: Optional[Literal["none"]] = None
"""
When not null, this overrides global both zone level and operation level
mitigation actions. This can serve as a quick way to disable schema validation
for the whole zone.
- `"none"` will skip running schema validation entirely for the request
"""

View file

@ -0,0 +1,28 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from typing import Optional
from typing_extensions import Literal, Required, TypedDict
__all__ = ["SettingUpdateParams"]
class SettingUpdateParams(TypedDict, total=False):
zone_id: Required[str]
"""Identifier."""
validation_default_mitigation_action: Required[Literal["none", "log", "block"]]
"""The default mitigation action used Mitigation actions are as follows:
- `"log"` - log request when request does not conform to schema
- `"block"` - deny access to the site when request does not conform to schema
- `"none"` - skip running schema validation
"""
validation_override_mitigation_action: Optional[Literal["none"]]
"""When set, this overrides both zone level and operation level mitigation actions.
- `"none"` - skip running schema validation entirely for the request
- `null` - clears any existing override
"""

View file

@ -0,0 +1,29 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from typing import Optional
from typing_extensions import Literal
from ..._models import BaseModel
__all__ = ["SettingUpdateResponse"]
class SettingUpdateResponse(BaseModel):
validation_default_mitigation_action: Literal["none", "log", "block"]
"""The default mitigation action used
Mitigation actions are as follows:
- `log` - log request when request does not conform to schema
- `block` - deny access to the site when request does not conform to schema
- `none` - skip running schema validation
"""
validation_override_mitigation_action: Optional[Literal["none"]] = None
"""
When not null, this overrides global both zone level and operation level
mitigation actions. This can serve as a quick way to disable schema validation
for the whole zone.
- `"none"` will skip running schema validation entirely for the request
"""

View file

@ -0,0 +1,12 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from .operation_list_params import OperationListParams as OperationListParams
from .operation_get_response import OperationGetResponse as OperationGetResponse
from .operation_list_response import OperationListResponse as OperationListResponse
from .operation_update_params import OperationUpdateParams as OperationUpdateParams
from .operation_delete_response import OperationDeleteResponse as OperationDeleteResponse
from .operation_update_response import OperationUpdateResponse as OperationUpdateResponse
from .operation_bulk_edit_params import OperationBulkEditParams as OperationBulkEditParams
from .operation_bulk_edit_response import OperationBulkEditResponse as OperationBulkEditResponse

View file

@ -0,0 +1,25 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from typing import Dict, Optional
from typing_extensions import Literal, Required, TypedDict
__all__ = ["OperationBulkEditParams", "Body"]
class OperationBulkEditParams(TypedDict, total=False):
zone_id: Required[str]
"""Identifier."""
body: Required[Dict[str, Body]]
class Body(TypedDict, total=False):
mitigation_action: Optional[Literal["none", "log", "block"]]
"""Mitigation actions are as follows:
- `log` - log request when request does not conform to schema _ `block` - deny
access to the site when request does not conform to schema _ `none` - skip
running schema validation \\** null - clears any existing per-operation setting
"""

View file

@ -0,0 +1,28 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from typing import Dict
from typing_extensions import Literal, TypeAlias
from ...._models import BaseModel
__all__ = ["OperationBulkEditResponse", "OperationBulkEditResponseItem"]
class OperationBulkEditResponseItem(BaseModel):
mitigation_action: Literal["log", "block", "none"]
"""
When set, this applies a mitigation action to this operation which supersedes a
global schema validation setting just for this operation
- `"log"` - log request when request does not conform to schema for this
operation
- `"block"` - deny access to the site when request does not conform to schema
for this operation
- `"none"` - will skip mitigation for this operation
"""
operation_id: str
"""UUID."""
OperationBulkEditResponse: TypeAlias = Dict[str, OperationBulkEditResponseItem]

View file

@ -0,0 +1,12 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from typing import Optional
from ...._models import BaseModel
__all__ = ["OperationDeleteResponse"]
class OperationDeleteResponse(BaseModel):
operation_id: Optional[str] = None
"""UUID."""

View file

@ -0,0 +1,24 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from typing_extensions import Literal
from ...._models import BaseModel
__all__ = ["OperationGetResponse"]
class OperationGetResponse(BaseModel):
mitigation_action: Literal["log", "block", "none"]
"""
When set, this applies a mitigation action to this operation which supersedes a
global schema validation setting just for this operation
- `"log"` - log request when request does not conform to schema for this
operation
- `"block"` - deny access to the site when request does not conform to schema
for this operation
- `"none"` - will skip mitigation for this operation
"""
operation_id: str
"""UUID."""

View file

@ -0,0 +1,18 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from typing_extensions import Required, TypedDict
__all__ = ["OperationListParams"]
class OperationListParams(TypedDict, total=False):
zone_id: Required[str]
"""Identifier."""
page: int
"""Page number of paginated results."""
per_page: int
"""Maximum number of results per page."""

View file

@ -0,0 +1,24 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from typing_extensions import Literal
from ...._models import BaseModel
__all__ = ["OperationListResponse"]
class OperationListResponse(BaseModel):
mitigation_action: Literal["log", "block", "none"]
"""
When set, this applies a mitigation action to this operation which supersedes a
global schema validation setting just for this operation
- `"log"` - log request when request does not conform to schema for this
operation
- `"block"` - deny access to the site when request does not conform to schema
for this operation
- `"none"` - will skip mitigation for this operation
"""
operation_id: str
"""UUID."""

View file

@ -0,0 +1,24 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from typing import Optional
from typing_extensions import Literal, Required, TypedDict
__all__ = ["OperationUpdateParams"]
class OperationUpdateParams(TypedDict, total=False):
zone_id: Required[str]
"""Identifier."""
mitigation_action: Required[Optional[Literal["log", "block", "none"]]]
"""When set, this applies a mitigation action to this operation
- `"log"` - log request when request does not conform to schema for this
operation
- `"block"` - deny access to the site when request does not conform to schema
for this operation
- `"none"` - will skip mitigation for this operation
- `null` - clears any mitigation action
"""

View file

@ -0,0 +1,24 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from typing_extensions import Literal
from ...._models import BaseModel
__all__ = ["OperationUpdateResponse"]
class OperationUpdateResponse(BaseModel):
mitigation_action: Literal["log", "block", "none"]
"""
When set, this applies a mitigation action to this operation which supersedes a
global schema validation setting just for this operation
- `"log"` - log request when request does not conform to schema for this
operation
- `"block"` - deny access to the site when request does not conform to schema
for this operation
- `"none"` - will skip mitigation for this operation
"""
operation_id: str
"""UUID."""

View file

@ -0,0 +1 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.

View file

@ -0,0 +1 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.

View file

@ -0,0 +1,529 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
import os
from typing import Any, cast
import pytest
from cloudflare import Cloudflare, AsyncCloudflare
from tests.utils import assert_matches_type
from cloudflare.pagination import SyncV4PagePaginationArray, AsyncV4PagePaginationArray
from cloudflare.types.schema_validation.settings import (
OperationGetResponse,
OperationListResponse,
OperationDeleteResponse,
OperationUpdateResponse,
OperationBulkEditResponse,
)
base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
class TestOperations:
parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
@parametrize
def test_method_update(self, client: Cloudflare) -> None:
operation = client.schema_validation.settings.operations.update(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
mitigation_action="block",
)
assert_matches_type(OperationUpdateResponse, operation, path=["response"])
@parametrize
def test_raw_response_update(self, client: Cloudflare) -> None:
response = client.schema_validation.settings.operations.with_raw_response.update(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
mitigation_action="block",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = response.parse()
assert_matches_type(OperationUpdateResponse, operation, path=["response"])
@parametrize
def test_streaming_response_update(self, client: Cloudflare) -> None:
with client.schema_validation.settings.operations.with_streaming_response.update(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
mitigation_action="block",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = response.parse()
assert_matches_type(OperationUpdateResponse, operation, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
def test_path_params_update(self, client: Cloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
client.schema_validation.settings.operations.with_raw_response.update(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="",
mitigation_action="block",
)
with pytest.raises(ValueError, match=r"Expected a non-empty value for `operation_id` but received ''"):
client.schema_validation.settings.operations.with_raw_response.update(
operation_id="",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
mitigation_action="block",
)
@parametrize
def test_method_list(self, client: Cloudflare) -> None:
operation = client.schema_validation.settings.operations.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(SyncV4PagePaginationArray[OperationListResponse], operation, path=["response"])
@parametrize
def test_method_list_with_all_params(self, client: Cloudflare) -> None:
operation = client.schema_validation.settings.operations.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
page=1,
per_page=5,
)
assert_matches_type(SyncV4PagePaginationArray[OperationListResponse], operation, path=["response"])
@parametrize
def test_raw_response_list(self, client: Cloudflare) -> None:
response = client.schema_validation.settings.operations.with_raw_response.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = response.parse()
assert_matches_type(SyncV4PagePaginationArray[OperationListResponse], operation, path=["response"])
@parametrize
def test_streaming_response_list(self, client: Cloudflare) -> None:
with client.schema_validation.settings.operations.with_streaming_response.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = response.parse()
assert_matches_type(SyncV4PagePaginationArray[OperationListResponse], operation, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
def test_path_params_list(self, client: Cloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
client.schema_validation.settings.operations.with_raw_response.list(
zone_id="",
)
@parametrize
def test_method_delete(self, client: Cloudflare) -> None:
operation = client.schema_validation.settings.operations.delete(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(OperationDeleteResponse, operation, path=["response"])
@parametrize
def test_raw_response_delete(self, client: Cloudflare) -> None:
response = client.schema_validation.settings.operations.with_raw_response.delete(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = response.parse()
assert_matches_type(OperationDeleteResponse, operation, path=["response"])
@parametrize
def test_streaming_response_delete(self, client: Cloudflare) -> None:
with client.schema_validation.settings.operations.with_streaming_response.delete(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = response.parse()
assert_matches_type(OperationDeleteResponse, operation, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
def test_path_params_delete(self, client: Cloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
client.schema_validation.settings.operations.with_raw_response.delete(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="",
)
with pytest.raises(ValueError, match=r"Expected a non-empty value for `operation_id` but received ''"):
client.schema_validation.settings.operations.with_raw_response.delete(
operation_id="",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
@parametrize
def test_method_bulk_edit(self, client: Cloudflare) -> None:
operation = client.schema_validation.settings.operations.bulk_edit(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
body={
"3818d821-5901-4147-a474-f5f5aec1d54e": {},
"b17c8043-99a0-4202-b7d9-8f7cdbee02cd": {},
},
)
assert_matches_type(OperationBulkEditResponse, operation, path=["response"])
@parametrize
def test_raw_response_bulk_edit(self, client: Cloudflare) -> None:
response = client.schema_validation.settings.operations.with_raw_response.bulk_edit(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
body={
"3818d821-5901-4147-a474-f5f5aec1d54e": {},
"b17c8043-99a0-4202-b7d9-8f7cdbee02cd": {},
},
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = response.parse()
assert_matches_type(OperationBulkEditResponse, operation, path=["response"])
@parametrize
def test_streaming_response_bulk_edit(self, client: Cloudflare) -> None:
with client.schema_validation.settings.operations.with_streaming_response.bulk_edit(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
body={
"3818d821-5901-4147-a474-f5f5aec1d54e": {},
"b17c8043-99a0-4202-b7d9-8f7cdbee02cd": {},
},
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = response.parse()
assert_matches_type(OperationBulkEditResponse, operation, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
def test_path_params_bulk_edit(self, client: Cloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
client.schema_validation.settings.operations.with_raw_response.bulk_edit(
zone_id="",
body={
"3818d821-5901-4147-a474-f5f5aec1d54e": {},
"b17c8043-99a0-4202-b7d9-8f7cdbee02cd": {},
},
)
@parametrize
def test_method_get(self, client: Cloudflare) -> None:
operation = client.schema_validation.settings.operations.get(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(OperationGetResponse, operation, path=["response"])
@parametrize
def test_raw_response_get(self, client: Cloudflare) -> None:
response = client.schema_validation.settings.operations.with_raw_response.get(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = response.parse()
assert_matches_type(OperationGetResponse, operation, path=["response"])
@parametrize
def test_streaming_response_get(self, client: Cloudflare) -> None:
with client.schema_validation.settings.operations.with_streaming_response.get(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = response.parse()
assert_matches_type(OperationGetResponse, operation, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
def test_path_params_get(self, client: Cloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
client.schema_validation.settings.operations.with_raw_response.get(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="",
)
with pytest.raises(ValueError, match=r"Expected a non-empty value for `operation_id` but received ''"):
client.schema_validation.settings.operations.with_raw_response.get(
operation_id="",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
class TestAsyncOperations:
parametrize = pytest.mark.parametrize("async_client", [False, True], indirect=True, ids=["loose", "strict"])
@parametrize
async def test_method_update(self, async_client: AsyncCloudflare) -> None:
operation = await async_client.schema_validation.settings.operations.update(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
mitigation_action="block",
)
assert_matches_type(OperationUpdateResponse, operation, path=["response"])
@parametrize
async def test_raw_response_update(self, async_client: AsyncCloudflare) -> None:
response = await async_client.schema_validation.settings.operations.with_raw_response.update(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
mitigation_action="block",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = await response.parse()
assert_matches_type(OperationUpdateResponse, operation, path=["response"])
@parametrize
async def test_streaming_response_update(self, async_client: AsyncCloudflare) -> None:
async with async_client.schema_validation.settings.operations.with_streaming_response.update(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
mitigation_action="block",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = await response.parse()
assert_matches_type(OperationUpdateResponse, operation, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
async def test_path_params_update(self, async_client: AsyncCloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
await async_client.schema_validation.settings.operations.with_raw_response.update(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="",
mitigation_action="block",
)
with pytest.raises(ValueError, match=r"Expected a non-empty value for `operation_id` but received ''"):
await async_client.schema_validation.settings.operations.with_raw_response.update(
operation_id="",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
mitigation_action="block",
)
@parametrize
async def test_method_list(self, async_client: AsyncCloudflare) -> None:
operation = await async_client.schema_validation.settings.operations.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(AsyncV4PagePaginationArray[OperationListResponse], operation, path=["response"])
@parametrize
async def test_method_list_with_all_params(self, async_client: AsyncCloudflare) -> None:
operation = await async_client.schema_validation.settings.operations.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
page=1,
per_page=5,
)
assert_matches_type(AsyncV4PagePaginationArray[OperationListResponse], operation, path=["response"])
@parametrize
async def test_raw_response_list(self, async_client: AsyncCloudflare) -> None:
response = await async_client.schema_validation.settings.operations.with_raw_response.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = await response.parse()
assert_matches_type(AsyncV4PagePaginationArray[OperationListResponse], operation, path=["response"])
@parametrize
async def test_streaming_response_list(self, async_client: AsyncCloudflare) -> None:
async with async_client.schema_validation.settings.operations.with_streaming_response.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = await response.parse()
assert_matches_type(AsyncV4PagePaginationArray[OperationListResponse], operation, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
async def test_path_params_list(self, async_client: AsyncCloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
await async_client.schema_validation.settings.operations.with_raw_response.list(
zone_id="",
)
@parametrize
async def test_method_delete(self, async_client: AsyncCloudflare) -> None:
operation = await async_client.schema_validation.settings.operations.delete(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(OperationDeleteResponse, operation, path=["response"])
@parametrize
async def test_raw_response_delete(self, async_client: AsyncCloudflare) -> None:
response = await async_client.schema_validation.settings.operations.with_raw_response.delete(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = await response.parse()
assert_matches_type(OperationDeleteResponse, operation, path=["response"])
@parametrize
async def test_streaming_response_delete(self, async_client: AsyncCloudflare) -> None:
async with async_client.schema_validation.settings.operations.with_streaming_response.delete(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = await response.parse()
assert_matches_type(OperationDeleteResponse, operation, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
async def test_path_params_delete(self, async_client: AsyncCloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
await async_client.schema_validation.settings.operations.with_raw_response.delete(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="",
)
with pytest.raises(ValueError, match=r"Expected a non-empty value for `operation_id` but received ''"):
await async_client.schema_validation.settings.operations.with_raw_response.delete(
operation_id="",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
@parametrize
async def test_method_bulk_edit(self, async_client: AsyncCloudflare) -> None:
operation = await async_client.schema_validation.settings.operations.bulk_edit(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
body={
"3818d821-5901-4147-a474-f5f5aec1d54e": {},
"b17c8043-99a0-4202-b7d9-8f7cdbee02cd": {},
},
)
assert_matches_type(OperationBulkEditResponse, operation, path=["response"])
@parametrize
async def test_raw_response_bulk_edit(self, async_client: AsyncCloudflare) -> None:
response = await async_client.schema_validation.settings.operations.with_raw_response.bulk_edit(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
body={
"3818d821-5901-4147-a474-f5f5aec1d54e": {},
"b17c8043-99a0-4202-b7d9-8f7cdbee02cd": {},
},
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = await response.parse()
assert_matches_type(OperationBulkEditResponse, operation, path=["response"])
@parametrize
async def test_streaming_response_bulk_edit(self, async_client: AsyncCloudflare) -> None:
async with async_client.schema_validation.settings.operations.with_streaming_response.bulk_edit(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
body={
"3818d821-5901-4147-a474-f5f5aec1d54e": {},
"b17c8043-99a0-4202-b7d9-8f7cdbee02cd": {},
},
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = await response.parse()
assert_matches_type(OperationBulkEditResponse, operation, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
async def test_path_params_bulk_edit(self, async_client: AsyncCloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
await async_client.schema_validation.settings.operations.with_raw_response.bulk_edit(
zone_id="",
body={
"3818d821-5901-4147-a474-f5f5aec1d54e": {},
"b17c8043-99a0-4202-b7d9-8f7cdbee02cd": {},
},
)
@parametrize
async def test_method_get(self, async_client: AsyncCloudflare) -> None:
operation = await async_client.schema_validation.settings.operations.get(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(OperationGetResponse, operation, path=["response"])
@parametrize
async def test_raw_response_get(self, async_client: AsyncCloudflare) -> None:
response = await async_client.schema_validation.settings.operations.with_raw_response.get(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = await response.parse()
assert_matches_type(OperationGetResponse, operation, path=["response"])
@parametrize
async def test_streaming_response_get(self, async_client: AsyncCloudflare) -> None:
async with async_client.schema_validation.settings.operations.with_streaming_response.get(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
operation = await response.parse()
assert_matches_type(OperationGetResponse, operation, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
async def test_path_params_get(self, async_client: AsyncCloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
await async_client.schema_validation.settings.operations.with_raw_response.get(
operation_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="",
)
with pytest.raises(ValueError, match=r"Expected a non-empty value for `operation_id` but received ''"):
await async_client.schema_validation.settings.operations.with_raw_response.get(
operation_id="",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)

View file

@ -0,0 +1,573 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
import os
from typing import Any, cast
import pytest
from cloudflare import Cloudflare, AsyncCloudflare
from tests.utils import assert_matches_type
from cloudflare.pagination import SyncV4PagePaginationArray, AsyncV4PagePaginationArray
from cloudflare.types.schema_validation import (
SchemaGetResponse,
SchemaEditResponse,
SchemaListResponse,
SchemaCreateResponse,
SchemaDeleteResponse,
)
base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
class TestSchemas:
parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
@parametrize
def test_method_create(self, client: Cloudflare) -> None:
schema = client.schema_validation.schemas.create(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
kind="openapi_v3",
name="petstore schema",
source="<schema file contents>",
)
assert_matches_type(SchemaCreateResponse, schema, path=["response"])
@parametrize
def test_method_create_with_all_params(self, client: Cloudflare) -> None:
schema = client.schema_validation.schemas.create(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
kind="openapi_v3",
name="petstore schema",
source="<schema file contents>",
validation_enabled=True,
)
assert_matches_type(SchemaCreateResponse, schema, path=["response"])
@parametrize
def test_raw_response_create(self, client: Cloudflare) -> None:
response = client.schema_validation.schemas.with_raw_response.create(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
kind="openapi_v3",
name="petstore schema",
source="<schema file contents>",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = response.parse()
assert_matches_type(SchemaCreateResponse, schema, path=["response"])
@parametrize
def test_streaming_response_create(self, client: Cloudflare) -> None:
with client.schema_validation.schemas.with_streaming_response.create(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
kind="openapi_v3",
name="petstore schema",
source="<schema file contents>",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = response.parse()
assert_matches_type(SchemaCreateResponse, schema, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
def test_path_params_create(self, client: Cloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
client.schema_validation.schemas.with_raw_response.create(
zone_id="",
kind="openapi_v3",
name="petstore schema",
source="<schema file contents>",
)
@parametrize
def test_method_list(self, client: Cloudflare) -> None:
schema = client.schema_validation.schemas.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(SyncV4PagePaginationArray[SchemaListResponse], schema, path=["response"])
@parametrize
def test_method_list_with_all_params(self, client: Cloudflare) -> None:
schema = client.schema_validation.schemas.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
omit_source=True,
page=1,
per_page=5,
validation_enabled=True,
)
assert_matches_type(SyncV4PagePaginationArray[SchemaListResponse], schema, path=["response"])
@parametrize
def test_raw_response_list(self, client: Cloudflare) -> None:
response = client.schema_validation.schemas.with_raw_response.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = response.parse()
assert_matches_type(SyncV4PagePaginationArray[SchemaListResponse], schema, path=["response"])
@parametrize
def test_streaming_response_list(self, client: Cloudflare) -> None:
with client.schema_validation.schemas.with_streaming_response.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = response.parse()
assert_matches_type(SyncV4PagePaginationArray[SchemaListResponse], schema, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
def test_path_params_list(self, client: Cloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
client.schema_validation.schemas.with_raw_response.list(
zone_id="",
)
@parametrize
def test_method_delete(self, client: Cloudflare) -> None:
schema = client.schema_validation.schemas.delete(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(SchemaDeleteResponse, schema, path=["response"])
@parametrize
def test_raw_response_delete(self, client: Cloudflare) -> None:
response = client.schema_validation.schemas.with_raw_response.delete(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = response.parse()
assert_matches_type(SchemaDeleteResponse, schema, path=["response"])
@parametrize
def test_streaming_response_delete(self, client: Cloudflare) -> None:
with client.schema_validation.schemas.with_streaming_response.delete(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = response.parse()
assert_matches_type(SchemaDeleteResponse, schema, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
def test_path_params_delete(self, client: Cloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
client.schema_validation.schemas.with_raw_response.delete(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="",
)
with pytest.raises(ValueError, match=r"Expected a non-empty value for `schema_id` but received ''"):
client.schema_validation.schemas.with_raw_response.delete(
schema_id="",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
@parametrize
def test_method_edit(self, client: Cloudflare) -> None:
schema = client.schema_validation.schemas.edit(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(SchemaEditResponse, schema, path=["response"])
@parametrize
def test_method_edit_with_all_params(self, client: Cloudflare) -> None:
schema = client.schema_validation.schemas.edit(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
validation_enabled=True,
)
assert_matches_type(SchemaEditResponse, schema, path=["response"])
@parametrize
def test_raw_response_edit(self, client: Cloudflare) -> None:
response = client.schema_validation.schemas.with_raw_response.edit(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = response.parse()
assert_matches_type(SchemaEditResponse, schema, path=["response"])
@parametrize
def test_streaming_response_edit(self, client: Cloudflare) -> None:
with client.schema_validation.schemas.with_streaming_response.edit(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = response.parse()
assert_matches_type(SchemaEditResponse, schema, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
def test_path_params_edit(self, client: Cloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
client.schema_validation.schemas.with_raw_response.edit(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="",
)
with pytest.raises(ValueError, match=r"Expected a non-empty value for `schema_id` but received ''"):
client.schema_validation.schemas.with_raw_response.edit(
schema_id="",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
@parametrize
def test_method_get(self, client: Cloudflare) -> None:
schema = client.schema_validation.schemas.get(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(SchemaGetResponse, schema, path=["response"])
@parametrize
def test_method_get_with_all_params(self, client: Cloudflare) -> None:
schema = client.schema_validation.schemas.get(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
omit_source=True,
)
assert_matches_type(SchemaGetResponse, schema, path=["response"])
@parametrize
def test_raw_response_get(self, client: Cloudflare) -> None:
response = client.schema_validation.schemas.with_raw_response.get(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = response.parse()
assert_matches_type(SchemaGetResponse, schema, path=["response"])
@parametrize
def test_streaming_response_get(self, client: Cloudflare) -> None:
with client.schema_validation.schemas.with_streaming_response.get(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = response.parse()
assert_matches_type(SchemaGetResponse, schema, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
def test_path_params_get(self, client: Cloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
client.schema_validation.schemas.with_raw_response.get(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="",
)
with pytest.raises(ValueError, match=r"Expected a non-empty value for `schema_id` but received ''"):
client.schema_validation.schemas.with_raw_response.get(
schema_id="",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
class TestAsyncSchemas:
parametrize = pytest.mark.parametrize("async_client", [False, True], indirect=True, ids=["loose", "strict"])
@parametrize
async def test_method_create(self, async_client: AsyncCloudflare) -> None:
schema = await async_client.schema_validation.schemas.create(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
kind="openapi_v3",
name="petstore schema",
source="<schema file contents>",
)
assert_matches_type(SchemaCreateResponse, schema, path=["response"])
@parametrize
async def test_method_create_with_all_params(self, async_client: AsyncCloudflare) -> None:
schema = await async_client.schema_validation.schemas.create(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
kind="openapi_v3",
name="petstore schema",
source="<schema file contents>",
validation_enabled=True,
)
assert_matches_type(SchemaCreateResponse, schema, path=["response"])
@parametrize
async def test_raw_response_create(self, async_client: AsyncCloudflare) -> None:
response = await async_client.schema_validation.schemas.with_raw_response.create(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
kind="openapi_v3",
name="petstore schema",
source="<schema file contents>",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = await response.parse()
assert_matches_type(SchemaCreateResponse, schema, path=["response"])
@parametrize
async def test_streaming_response_create(self, async_client: AsyncCloudflare) -> None:
async with async_client.schema_validation.schemas.with_streaming_response.create(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
kind="openapi_v3",
name="petstore schema",
source="<schema file contents>",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = await response.parse()
assert_matches_type(SchemaCreateResponse, schema, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
async def test_path_params_create(self, async_client: AsyncCloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
await async_client.schema_validation.schemas.with_raw_response.create(
zone_id="",
kind="openapi_v3",
name="petstore schema",
source="<schema file contents>",
)
@parametrize
async def test_method_list(self, async_client: AsyncCloudflare) -> None:
schema = await async_client.schema_validation.schemas.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(AsyncV4PagePaginationArray[SchemaListResponse], schema, path=["response"])
@parametrize
async def test_method_list_with_all_params(self, async_client: AsyncCloudflare) -> None:
schema = await async_client.schema_validation.schemas.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
omit_source=True,
page=1,
per_page=5,
validation_enabled=True,
)
assert_matches_type(AsyncV4PagePaginationArray[SchemaListResponse], schema, path=["response"])
@parametrize
async def test_raw_response_list(self, async_client: AsyncCloudflare) -> None:
response = await async_client.schema_validation.schemas.with_raw_response.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = await response.parse()
assert_matches_type(AsyncV4PagePaginationArray[SchemaListResponse], schema, path=["response"])
@parametrize
async def test_streaming_response_list(self, async_client: AsyncCloudflare) -> None:
async with async_client.schema_validation.schemas.with_streaming_response.list(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = await response.parse()
assert_matches_type(AsyncV4PagePaginationArray[SchemaListResponse], schema, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
async def test_path_params_list(self, async_client: AsyncCloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
await async_client.schema_validation.schemas.with_raw_response.list(
zone_id="",
)
@parametrize
async def test_method_delete(self, async_client: AsyncCloudflare) -> None:
schema = await async_client.schema_validation.schemas.delete(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(SchemaDeleteResponse, schema, path=["response"])
@parametrize
async def test_raw_response_delete(self, async_client: AsyncCloudflare) -> None:
response = await async_client.schema_validation.schemas.with_raw_response.delete(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = await response.parse()
assert_matches_type(SchemaDeleteResponse, schema, path=["response"])
@parametrize
async def test_streaming_response_delete(self, async_client: AsyncCloudflare) -> None:
async with async_client.schema_validation.schemas.with_streaming_response.delete(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = await response.parse()
assert_matches_type(SchemaDeleteResponse, schema, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
async def test_path_params_delete(self, async_client: AsyncCloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
await async_client.schema_validation.schemas.with_raw_response.delete(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="",
)
with pytest.raises(ValueError, match=r"Expected a non-empty value for `schema_id` but received ''"):
await async_client.schema_validation.schemas.with_raw_response.delete(
schema_id="",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
@parametrize
async def test_method_edit(self, async_client: AsyncCloudflare) -> None:
schema = await async_client.schema_validation.schemas.edit(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(SchemaEditResponse, schema, path=["response"])
@parametrize
async def test_method_edit_with_all_params(self, async_client: AsyncCloudflare) -> None:
schema = await async_client.schema_validation.schemas.edit(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
validation_enabled=True,
)
assert_matches_type(SchemaEditResponse, schema, path=["response"])
@parametrize
async def test_raw_response_edit(self, async_client: AsyncCloudflare) -> None:
response = await async_client.schema_validation.schemas.with_raw_response.edit(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = await response.parse()
assert_matches_type(SchemaEditResponse, schema, path=["response"])
@parametrize
async def test_streaming_response_edit(self, async_client: AsyncCloudflare) -> None:
async with async_client.schema_validation.schemas.with_streaming_response.edit(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = await response.parse()
assert_matches_type(SchemaEditResponse, schema, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
async def test_path_params_edit(self, async_client: AsyncCloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
await async_client.schema_validation.schemas.with_raw_response.edit(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="",
)
with pytest.raises(ValueError, match=r"Expected a non-empty value for `schema_id` but received ''"):
await async_client.schema_validation.schemas.with_raw_response.edit(
schema_id="",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
@parametrize
async def test_method_get(self, async_client: AsyncCloudflare) -> None:
schema = await async_client.schema_validation.schemas.get(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(SchemaGetResponse, schema, path=["response"])
@parametrize
async def test_method_get_with_all_params(self, async_client: AsyncCloudflare) -> None:
schema = await async_client.schema_validation.schemas.get(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
omit_source=True,
)
assert_matches_type(SchemaGetResponse, schema, path=["response"])
@parametrize
async def test_raw_response_get(self, async_client: AsyncCloudflare) -> None:
response = await async_client.schema_validation.schemas.with_raw_response.get(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = await response.parse()
assert_matches_type(SchemaGetResponse, schema, path=["response"])
@parametrize
async def test_streaming_response_get(self, async_client: AsyncCloudflare) -> None:
async with async_client.schema_validation.schemas.with_streaming_response.get(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
schema = await response.parse()
assert_matches_type(SchemaGetResponse, schema, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
async def test_path_params_get(self, async_client: AsyncCloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
await async_client.schema_validation.schemas.with_raw_response.get(
schema_id="f174e90a-fafe-4643-bbbc-4a0ed4fc8415",
zone_id="",
)
with pytest.raises(ValueError, match=r"Expected a non-empty value for `schema_id` but received ''"):
await async_client.schema_validation.schemas.with_raw_response.get(
schema_id="",
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)

View file

@ -0,0 +1,298 @@
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
import os
from typing import Any, cast
import pytest
from cloudflare import Cloudflare, AsyncCloudflare
from tests.utils import assert_matches_type
from cloudflare.types.schema_validation import (
SettingGetResponse,
SettingEditResponse,
SettingUpdateResponse,
)
base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010")
class TestSettings:
parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"])
@parametrize
def test_method_update(self, client: Cloudflare) -> None:
setting = client.schema_validation.settings.update(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
validation_default_mitigation_action="block",
)
assert_matches_type(SettingUpdateResponse, setting, path=["response"])
@parametrize
def test_method_update_with_all_params(self, client: Cloudflare) -> None:
setting = client.schema_validation.settings.update(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
validation_default_mitigation_action="block",
validation_override_mitigation_action="none",
)
assert_matches_type(SettingUpdateResponse, setting, path=["response"])
@parametrize
def test_raw_response_update(self, client: Cloudflare) -> None:
response = client.schema_validation.settings.with_raw_response.update(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
validation_default_mitigation_action="block",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
setting = response.parse()
assert_matches_type(SettingUpdateResponse, setting, path=["response"])
@parametrize
def test_streaming_response_update(self, client: Cloudflare) -> None:
with client.schema_validation.settings.with_streaming_response.update(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
validation_default_mitigation_action="block",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
setting = response.parse()
assert_matches_type(SettingUpdateResponse, setting, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
def test_path_params_update(self, client: Cloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
client.schema_validation.settings.with_raw_response.update(
zone_id="",
validation_default_mitigation_action="block",
)
@parametrize
def test_method_edit(self, client: Cloudflare) -> None:
setting = client.schema_validation.settings.edit(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(SettingEditResponse, setting, path=["response"])
@parametrize
def test_method_edit_with_all_params(self, client: Cloudflare) -> None:
setting = client.schema_validation.settings.edit(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
validation_default_mitigation_action="block",
validation_override_mitigation_action="none",
)
assert_matches_type(SettingEditResponse, setting, path=["response"])
@parametrize
def test_raw_response_edit(self, client: Cloudflare) -> None:
response = client.schema_validation.settings.with_raw_response.edit(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
setting = response.parse()
assert_matches_type(SettingEditResponse, setting, path=["response"])
@parametrize
def test_streaming_response_edit(self, client: Cloudflare) -> None:
with client.schema_validation.settings.with_streaming_response.edit(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
setting = response.parse()
assert_matches_type(SettingEditResponse, setting, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
def test_path_params_edit(self, client: Cloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
client.schema_validation.settings.with_raw_response.edit(
zone_id="",
)
@parametrize
def test_method_get(self, client: Cloudflare) -> None:
setting = client.schema_validation.settings.get(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(SettingGetResponse, setting, path=["response"])
@parametrize
def test_raw_response_get(self, client: Cloudflare) -> None:
response = client.schema_validation.settings.with_raw_response.get(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
setting = response.parse()
assert_matches_type(SettingGetResponse, setting, path=["response"])
@parametrize
def test_streaming_response_get(self, client: Cloudflare) -> None:
with client.schema_validation.settings.with_streaming_response.get(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
setting = response.parse()
assert_matches_type(SettingGetResponse, setting, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
def test_path_params_get(self, client: Cloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
client.schema_validation.settings.with_raw_response.get(
zone_id="",
)
class TestAsyncSettings:
parametrize = pytest.mark.parametrize("async_client", [False, True], indirect=True, ids=["loose", "strict"])
@parametrize
async def test_method_update(self, async_client: AsyncCloudflare) -> None:
setting = await async_client.schema_validation.settings.update(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
validation_default_mitigation_action="block",
)
assert_matches_type(SettingUpdateResponse, setting, path=["response"])
@parametrize
async def test_method_update_with_all_params(self, async_client: AsyncCloudflare) -> None:
setting = await async_client.schema_validation.settings.update(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
validation_default_mitigation_action="block",
validation_override_mitigation_action="none",
)
assert_matches_type(SettingUpdateResponse, setting, path=["response"])
@parametrize
async def test_raw_response_update(self, async_client: AsyncCloudflare) -> None:
response = await async_client.schema_validation.settings.with_raw_response.update(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
validation_default_mitigation_action="block",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
setting = await response.parse()
assert_matches_type(SettingUpdateResponse, setting, path=["response"])
@parametrize
async def test_streaming_response_update(self, async_client: AsyncCloudflare) -> None:
async with async_client.schema_validation.settings.with_streaming_response.update(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
validation_default_mitigation_action="block",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
setting = await response.parse()
assert_matches_type(SettingUpdateResponse, setting, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
async def test_path_params_update(self, async_client: AsyncCloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
await async_client.schema_validation.settings.with_raw_response.update(
zone_id="",
validation_default_mitigation_action="block",
)
@parametrize
async def test_method_edit(self, async_client: AsyncCloudflare) -> None:
setting = await async_client.schema_validation.settings.edit(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(SettingEditResponse, setting, path=["response"])
@parametrize
async def test_method_edit_with_all_params(self, async_client: AsyncCloudflare) -> None:
setting = await async_client.schema_validation.settings.edit(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
validation_default_mitigation_action="block",
validation_override_mitigation_action="none",
)
assert_matches_type(SettingEditResponse, setting, path=["response"])
@parametrize
async def test_raw_response_edit(self, async_client: AsyncCloudflare) -> None:
response = await async_client.schema_validation.settings.with_raw_response.edit(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
setting = await response.parse()
assert_matches_type(SettingEditResponse, setting, path=["response"])
@parametrize
async def test_streaming_response_edit(self, async_client: AsyncCloudflare) -> None:
async with async_client.schema_validation.settings.with_streaming_response.edit(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
setting = await response.parse()
assert_matches_type(SettingEditResponse, setting, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
async def test_path_params_edit(self, async_client: AsyncCloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
await async_client.schema_validation.settings.with_raw_response.edit(
zone_id="",
)
@parametrize
async def test_method_get(self, async_client: AsyncCloudflare) -> None:
setting = await async_client.schema_validation.settings.get(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert_matches_type(SettingGetResponse, setting, path=["response"])
@parametrize
async def test_raw_response_get(self, async_client: AsyncCloudflare) -> None:
response = await async_client.schema_validation.settings.with_raw_response.get(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
)
assert response.is_closed is True
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
setting = await response.parse()
assert_matches_type(SettingGetResponse, setting, path=["response"])
@parametrize
async def test_streaming_response_get(self, async_client: AsyncCloudflare) -> None:
async with async_client.schema_validation.settings.with_streaming_response.get(
zone_id="023e105f4ecef8ad9ca31a8372d0c353",
) as response:
assert not response.is_closed
assert response.http_request.headers.get("X-Stainless-Lang") == "python"
setting = await response.parse()
assert_matches_type(SettingGetResponse, setting, path=["response"])
assert cast(Any, response.is_closed) is True
@parametrize
async def test_path_params_get(self, async_client: AsyncCloudflare) -> None:
with pytest.raises(ValueError, match=r"Expected a non-empty value for `zone_id` but received ''"):
await async_client.schema_validation.settings.with_raw_response.get(
zone_id="",
)