Define Control Points

Aperture Python SDK can be used to define feature control points within a Python service.

Run the command below to install the SDK:

pip install aperture-py

The next step is to create an Aperture Client instance, for which, the address of the organization created in Aperture Cloud and API key are needed. You can locate both these details by clicking on the Aperture tab in the sidebar menu of Aperture Cloud.

API Key

You can create an API key for your project in the Aperture Cloud UI. For detailed instructions on locating API Keys, please refer to the API Keys section.

Python Async

from aperture_sdk.client_async import ApertureClientAsync, FlowParams


agent_address = os.getenv("APERTURE_AGENT_ADDRESS", default_agent_address)
api_key = os.getenv("APERTURE_API_KEY", "")
insecure = os.getenv("APERTURE_AGENT_INSECURE", "true").lower() == "true"

aperture_client = ApertureClientAsync.new_client(
    address=agent_address, insecure=insecure, api_key=api_key
)

Python Sync

from aperture_sdk.client import ApertureClient, FlowParams


agent_address = os.getenv("APERTURE_AGENT_ADDRESS", default_agent_address)
api_key = os.getenv("APERTURE_API_KEY", "")
insecure = os.getenv("APERTURE_AGENT_INSECURE", "true").lower() == "true"

aperture_client = ApertureClient.new_client(
    address=agent_address, insecure=insecure, api_key=api_key
)

The created instance can then be used to start a flow:

Python Async

# business logic produces labels
    labels = {
        "user_id": "some_user_id",
        "user_tier": "gold",
        "priority": "100",
    }
    flow_params = FlowParams(
        check_timeout=timedelta(seconds=200),
        explicit_labels=labels,
    )
    # start_flow performs a flowcontrol.v1.Check call to Aperture Agent.
    # It returns a Flow or raises an error if any.
    flow = await aperture_client.start_flow(
        control_point="AwesomeFeature",
        params=flow_params,
    )

    # Check if flow check was successful.
    if not flow.success:
        logger.info("Flow check failed - will fail-open")

    # See whether flow was accepted by Aperture Agent.
    if flow.should_run():
        # do actual work
        pass
    else:
        # handle flow rejection by Aperture Agent
        flow.set_status(FlowStatus.Error)

    res = await flow.end()
    if res.get_error():
        logger.error("Error: {}".format(res.get_error()))
    elif res.get_flow_end_response():
        logger.info("Flow End Response: {}".format(res.get_flow_end_response()))

    # Simulate work being done
    await asyncio.sleep(2)
    return "", 202

Python Sync

labels = {
        "userId": "some_user_id",
        "userTier": "gold",
        "priority": "100",
    }
    flow_params = FlowParams(
        check_timeout=timedelta(seconds=400),
        explicit_labels=labels,
    )
    flow = aperture_client.start_flow(
        control_point="rate-limiting-feature",
        params=flow_params,
    )

    if flow.should_run():
        print("Request accepted. Processing...")
    else:
        print("Request rate-limited. Try again later.")
        flow.set_status()

    flow.end()
    #Simulate work being done
    await asyncio.sleep(1)

The above code snippet is making start_flow calls to Aperture. For this call, it is important to specify the control point (AwesomeFeature in the example) and FlowParams that will be aligned with the policy created in Aperture Cloud. For request prioritization use cases, it's important to set a higher gRPC deadline. This parameter specifies the maximum duration a request can remain in the queue. For each flow that is started, a should_run decision is made, determining whether to allow the request into the system or to rate limit it. It is important to make the end call made after processing each request, to send telemetry data that would provide granular visibility for each flow.

You can also use the flow as a context manager:

Python

flow_params = FlowParams(
        explicit_labels=labels,
        check_timeout=timedelta(seconds=200),
    )

    with await aperture_client.start_flow(
        control_point="AwesomeFeature",
        params=flow_params,
    ) as flow:
        if flow.should_run():
            # do actual work
            # if you do not call flow.end() explicitly, it will be called automatically
            # when the context manager exits - with the status of the flow
            # depending on whether an error was raised or not
            pass

Additionally, you can decorate any function with aperture client. This will skip running the function if the flow is rejected by Aperture Agent. This might be helpful to handle specific routes in your service.

Python

flow_params = FlowParams(
    check_timeout=timedelta(seconds=200),
)


@app.get("/super")
@aperture_client.decorate(
    "awesomeFeature", params=flow_params, on_reject=lambda: ("Flow was rejected", 503)
)
async def super_handler():
    # Simulate work being done
    await asyncio.sleep(2)
    return "", 202

For more context on using the Aperture Python SDK to set feature control points, refer to the example app available in the repository.