tessl/pypi-azure-mgmt-apimanagement
Microsoft Azure API Management Client Library for Python providing comprehensive SDK functionality for managing Azure API Management services.
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
policy-management.mddocs/
Policy Management
Comprehensive policy system for implementing cross-cutting concerns including authentication, rate limiting, transformation, caching, and routing. Supports policies at global, product, API, and operation levels with reusable policy fragments and validation capabilities.
Capabilities
Global Policy Management
Service-level global policies that apply to all APIs and operations within the API Management instance.
class PolicyOperations:
def get(self, resource_group_name: str, service_name: str, policy_id: str, format: Optional[str] = None, **kwargs) -> PolicyContract: ...
def create_or_update(self, resource_group_name: str, service_name: str, policy_id: str, parameters: PolicyContract, if_match: Optional[str] = None, **kwargs) -> PolicyContract: ...
def delete(self, resource_group_name: str, service_name: str, policy_id: str, if_match: str, **kwargs) -> None: ...
def list_by_service(self, resource_group_name: str, service_name: str, **kwargs) -> PolicyCollection: ...
def get_entity_tag(self, resource_group_name: str, service_name: str, policy_id: str, **kwargs) -> bool: ...Policy Fragment Management
Reusable policy fragments that can be referenced across multiple policies for better maintainability and consistency.
class PolicyFragmentOperations:
def get(self, resource_group_name: str, service_name: str, policy_fragment_id: str, format: Optional[str] = None, **kwargs) -> PolicyFragmentContract: ...
def begin_create_or_update(self, resource_group_name: str, service_name: str, policy_fragment_id: str, parameters: PolicyFragmentContract, if_match: Optional[str] = None, **kwargs) -> LROPoller[PolicyFragmentContract]: ...
def begin_delete(self, resource_group_name: str, service_name: str, policy_fragment_id: str, if_match: str, **kwargs) -> LROPoller[None]: ...
def list_by_service(self, resource_group_name: str, service_name: str, filter: Optional[str] = None, top: Optional[int] = None, skip: Optional[int] = None, **kwargs) -> ItemPaged[PolicyFragmentContract]: ...
def get_entity_tag(self, resource_group_name: str, service_name: str, policy_fragment_id: str, **kwargs) -> bool: ...
def list_references(self, resource_group_name: str, service_name: str, policy_fragment_id: str, top: Optional[int] = None, skip: Optional[int] = None, **kwargs) -> ItemPaged[ResourceCollection]: ...Policy Description Operations
Management of policy descriptions and documentation for available policy types.
class PolicyDescriptionOperations:
def list_by_service(self, resource_group_name: str, service_name: str, scope: Optional[str] = None, **kwargs) -> PolicyDescriptionCollection: ...Policy Restriction Management
Policy validation and restriction enforcement to ensure policy compliance and security.
class PolicyRestrictionOperations:
def get(self, resource_group_name: str, service_name: str, policy_restriction_id: str, **kwargs) -> PolicyRestrictionContract: ...
def update(self, resource_group_name: str, service_name: str, policy_restriction_id: str, if_match: str, parameters: PolicyRestrictionContract, **kwargs) -> PolicyRestrictionContract: ...
def list_by_service(self, resource_group_name: str, service_name: str, **kwargs) -> PolicyRestrictionCollection: ...
def get_entity_tag(self, resource_group_name: str, service_name: str, policy_restriction_id: str, **kwargs) -> bool: ...
class PolicyRestrictionValidationsOperations:
def by_service_and_policy_id(self, resource_group_name: str, service_name: str, policy_restriction_id: str, parameters: PolicyRestrictionValidationParameters, **kwargs) -> PolicyRestrictionValidationResult: ...All Policies Operations
Comprehensive policy management across all scopes (global, product, API, operation).
class AllPoliciesOperations:
def list_by_service(self, resource_group_name: str, service_name: str, **kwargs) -> AllPoliciesCollection: ...API-Level Policy Operations
Policy management specific to individual APIs, inherited by all operations within the API unless overridden.
class ApiPolicyOperations:
def get(self, resource_group_name: str, service_name: str, api_id: str, policy_id: str, format: Optional[str] = None, **kwargs) -> PolicyContract: ...
def create_or_update(self, resource_group_name: str, service_name: str, api_id: str, policy_id: str, parameters: PolicyContract, if_match: Optional[str] = None, **kwargs) -> PolicyContract: ...
def delete(self, resource_group_name: str, service_name: str, api_id: str, policy_id: str, if_match: str, **kwargs) -> None: ...
def list_by_api(self, resource_group_name: str, service_name: str, api_id: str, **kwargs) -> PolicyCollection: ...
def get_entity_tag(self, resource_group_name: str, service_name: str, api_id: str, policy_id: str, **kwargs) -> bool: ...Operation-Level Policy Operations
Fine-grained policy management for individual API operations with the highest precedence in the policy hierarchy.
class ApiOperationPolicyOperations:
def get(self, resource_group_name: str, service_name: str, api_id: str, operation_id: str, policy_id: str, format: Optional[str] = None, **kwargs) -> PolicyContract: ...
def create_or_update(self, resource_group_name: str, service_name: str, api_id: str, operation_id: str, policy_id: str, parameters: PolicyContract, if_match: Optional[str] = None, **kwargs) -> PolicyContract: ...
def delete(self, resource_group_name: str, service_name: str, api_id: str, operation_id: str, policy_id: str, if_match: str, **kwargs) -> None: ...
def list_by_operation(self, resource_group_name: str, service_name: str, api_id: str, operation_id: str, **kwargs) -> PolicyCollection: ...
def get_entity_tag(self, resource_group_name: str, service_name: str, api_id: str, operation_id: str, policy_id: str, **kwargs) -> bool: ...Product-Level Policy Operations
Policy management for products, applying to all APIs within the product scope.
class ProductPolicyOperations:
def get(self, resource_group_name: str, service_name: str, product_id: str, policy_id: str, format: Optional[str] = None, **kwargs) -> PolicyContract: ...
def create_or_update(self, resource_group_name: str, service_name: str, product_id: str, policy_id: str, parameters: PolicyContract, if_match: Optional[str] = None, **kwargs) -> PolicyContract: ...
def delete(self, resource_group_name: str, service_name: str, product_id: str, policy_id: str, if_match: str, **kwargs) -> None: ...
def list_by_product(self, resource_group_name: str, service_name: str, product_id: str, **kwargs) -> PolicyCollection: ...
def get_entity_tag(self, resource_group_name: str, service_name: str, product_id: str, policy_id: str, **kwargs) -> bool: ...Usage Examples
Creating a Rate Limiting Policy
from azure.mgmt.apimanagement.models import PolicyContract
# Define rate limiting policy XML
rate_limit_policy = """
<policies>
<inbound>
<rate-limit calls="100" renewal-period="60" />
<quota calls="10000" renewal-period="604800" />
<base />
</inbound>
<backend>
<base />
</backend>
<outbound>
<base />
</outbound>
<on-error>
<base />
</on-error>
</policies>
"""
# Create policy contract
policy = PolicyContract(
value=rate_limit_policy,
format="xml"
)
# Apply policy at global level
global_policy = client.policy.create_or_update(
resource_group_name="my-rg",
service_name="my-apim",
policy_id="policy",
parameters=policy
)
print("Global rate limiting policy applied")Creating Authentication Policy
# JWT validation policy
jwt_policy = """
<policies>
<inbound>
<validate-jwt header-name="Authorization" failed-validation-httpcode="401" failed-validation-error-message="Unauthorized">
<openid-config url="https://login.microsoftonline.com/{tenant-id}/.well-known/openid_configuration" />
<required-claims>
<claim name="aud">
<value>{client-id}</value>
</claim>
</required-claims>
</validate-jwt>
<base />
</inbound>
<backend>
<base />
</backend>
<outbound>
<base />
</outbound>
<on-error>
<base />
</on-error>
</policies>
"""
# Apply to specific API
api_policy = PolicyContract(
value=jwt_policy,
format="xml"
)
api_auth_policy = client.api_policy.create_or_update(
resource_group_name="my-rg",
service_name="my-apim",
api_id="secure-api",
policy_id="policy",
parameters=api_policy
)
print("JWT authentication policy applied to API")Creating Reusable Policy Fragment
from azure.mgmt.apimanagement.models import PolicyFragmentContract
# CORS policy fragment
cors_fragment = """
<fragment>
<cors allow-credentials="true">
<allowed-origins>
<origin>https://www.contoso.com</origin>
<origin>https://app.contoso.com</origin>
</allowed-origins>
<allowed-methods preflight-result-max-age="300">
<method>GET</method>
<method>POST</method>
<method>PUT</method>
<method>DELETE</method>
<method>OPTIONS</method>
</allowed-methods>
<allowed-headers>
<header>Content-Type</header>
<header>Authorization</header>
</allowed-headers>
</cors>
</fragment>
"""
# Create policy fragment
fragment = PolicyFragmentContract(
value=cors_fragment,
description="Standard CORS policy for Contoso applications",
format="xml"
)
cors_fragment = client.policy_fragment.begin_create_or_update(
resource_group_name="my-rg",
service_name="my-apim",
policy_fragment_id="cors-fragment",
parameters=fragment
).result()
print(f"Policy fragment created: {cors_fragment.name}")Using Policy Fragment in Policy
# Policy that references the fragment
policy_with_fragment = """
<policies>
<inbound>
<include-fragment fragment-id="cors-fragment" />
<rate-limit calls="1000" renewal-period="60" />
<base />
</inbound>
<backend>
<base />
</backend>
<outbound>
<base />
</outbound>
<on-error>
<base />
</on-error>
</policies>
"""
# Apply policy using fragment
policy_with_fragment_contract = PolicyContract(
value=policy_with_fragment,
format="xml"
)
api_policy = client.api_policy.create_or_update(
resource_group_name="my-rg",
service_name="my-apim",
api_id="my-api",
policy_id="policy",
parameters=policy_with_fragment_contract
)Transformation Policy Example
# Request/response transformation policy
transformation_policy = """
<policies>
<inbound>
<set-header name="X-Custom-Header" exists-action="override">
<value>@(context.Request.Headers.GetValueOrDefault("User-Agent",""))</value>
</set-header>
<rewrite-uri template="/v2/{path-param}" />
<set-query-parameter name="api-version" exists-action="override">
<value>2.0</value>
</set-query-parameter>
<json-to-xml apply="always" consider-accept-header="false" />
<base />
</inbound>
<backend>
<retry condition="@(context.Response.StatusCode >= 500)" count="3" interval="1" />
<base />
</backend>
<outbound>
<xml-to-json kind="direct" apply="always" consider-accept-header="false" />
<find-and-replace from="oldValue" to="newValue" />
<base />
</outbound>
<on-error>
<set-status code="500" reason="Internal Server Error" />
<set-body>@{
return new JObject(
new JProperty("error", new JObject(
new JProperty("code", "InternalError"),
new JProperty("message", context.LastError.Message)
))
).ToString();
}</set-body>
<base />
</on-error>
</policies>
"""
transformation_policy_contract = PolicyContract(
value=transformation_policy,
format="xml"
)
operation_policy = client.api_operation_policy.create_or_update(
resource_group_name="my-rg",
service_name="my-apim",
api_id="transformation-api",
operation_id="transform-operation",
policy_id="policy",
parameters=transformation_policy_contract
)Types
Core Policy Types
class PolicyContract:
id: Optional[str]
name: Optional[str]
type: Optional[str]
properties: Optional[PolicyContractProperties]
class PolicyContractProperties:
value: str
format: Optional[PolicyContentFormat]
class PolicyFragmentContract:
id: Optional[str]
name: Optional[str]
type: Optional[str]
properties: Optional[PolicyFragmentContractProperties]
class PolicyFragmentContractProperties:
value: str
description: Optional[str]
format: Optional[PolicyContentFormat]
class PolicyCollection:
value: Optional[List[PolicyContract]]
next_link: Optional[str]
class PolicyDescriptionContract:
id: Optional[str]
name: Optional[str]
type: Optional[str]
properties: Optional[PolicyDescriptionContractProperties]
class PolicyDescriptionContractProperties:
display_name: Optional[str]
description: Optional[str]
scope: Optional[PolicyScopeContract]
class PolicyRestrictionContract:
id: Optional[str]
name: Optional[str]
type: Optional[str]
properties: Optional[PolicyRestrictionContractProperties]
class PolicyRestrictionContractProperties:
scope: Optional[str]
selector: Optional[str]
value: Optional[str]Policy Enums
class PolicyContentFormat:
XML = "xml"
XML_LINK = "xml-link"
RAWXML = "rawxml"
RAWXML_LINK = "rawxml-link"
class PolicyScopeContract:
TENANT = "Tenant"
PRODUCT = "Product"
API = "Api"
OPERATION = "Operation"
ALL = "All"
class PolicyExportFormat:
XML = "xml"
RAWXML = "rawxml"Common Policy Examples
Authentication Policies
- API Key:
<check-header name="Ocp-Apim-Subscription-Key" failed-check-httpcode="401" /> - JWT Validation:
<validate-jwt header-name="Authorization"> - OAuth 2.0:
<validate-jwt header-name="Authorization" require-scheme="Bearer"> - Certificate:
<authentication-certificate thumbprint="{thumbprint}" />
Rate Limiting Policies
- Rate Limit:
<rate-limit calls="100" renewal-period="60" /> - Quota:
<quota calls="10000" renewal-period="604800" /> - Rate Limit by Key:
<rate-limit-by-key calls="100" renewal-period="60" counter-key="@(context.Request.IpAddress)" />
Transformation Policies
- Set Header:
<set-header name="X-Custom-Header" exists-action="override"> - Rewrite URI:
<rewrite-uri template="/v2/users/{id}" /> - JSON to XML:
<json-to-xml apply="always" /> - Find and Replace:
<find-and-replace from="oldtext" to="newtext" />
Backend Policies
- Set Backend Service:
<set-backend-service base-url="https://backend.contoso.com" /> - Retry:
<retry condition="@(context.Response.StatusCode >= 500)" count="3" /> - Timeout:
<forward-request timeout="60" /> - Load Balancer:
<set-backend-service backend-id="load-balanced-backend" />
Install with Tessl CLI
npx tessl i tessl/pypi-azure-mgmt-apimanagement