Schematic Python Library

The Schematic Python Library provides convenient access to the Schematic API from applications written in Python.

The library includes type definitions for all request and response fields, and offers both synchronous and asynchronous clients powered by httpx.

Installation and Setup

1. Add schematichq to your project's build file:

pip install schematichq
# or
poetry add schematichq

2. Issue an API key for the appropriate environment using the Schematic app.

3. Using this secret key, initialize a client in your application:

fromschematic.clientimportSchematicclient=Schematic("YOUR_API_KEY")

Async Client

The SDK exports an async client for non-blocking API calls with automatic background event processing. The async client features lazy initialization - you can start using it immediately without manual setup.

Simple Usage (Lazy Initialization)

The easiest way to use the async client - just create and use it directly:

importasynciofromschematic.clientimportAsyncSchematicasyncdefmain():
# Create client - no initialize() needed!client=AsyncSchematic("YOUR_API_KEY")
# Use immediately - auto-initializes on first callis_enabled=awaitclient.check_flag(
"new-feature",
company={"id": "company-123"},
user={"id": "user-456"}
)
ifis_enabled:
print("New feature is enabled!")
# Track usageawaitclient.track(
event="feature-used",
company={"id": "company-123"},
user={"id": "user-456"}
)
# Always shutdown when doneawaitclient.shutdown()
asyncio.run(main())

Context Manager (Recommended)

Use the async client as a context manager for automatic lifecycle management:

importasynciofromschematic.clientimportAsyncSchematicasyncdefmain():
asyncwithAsyncSchematic("YOUR_API_KEY") asclient:
# Client auto-initializes and will auto-shutdownis_enabled=awaitclient.check_flag(
"feature-flag",
company={"id": "company-123"}
)
awaitclient.identify(
keys={"id": "company-123"},
name="Acme Corp"
)
# Automatic cleanup on exitasyncio.run(main())

Production Usage (Explicit Control)

For production applications that need precise control over initialization timing:

importasynciofromschematic.clientimportAsyncSchematic# Web application exampleclient=AsyncSchematic("YOUR_API_KEY")
asyncdefstartup():
"""Call during application startup"""awaitclient.initialize() # Start background tasks nowprint("Schematic client ready")
asyncdefshutdown():
"""Call during application shutdown"""awaitclient.shutdown() # Stop background tasks and flush eventsprint("Schematic client stopped")
asyncdefhandle_request():
"""Handle individual requests"""# Client is already initialized - this will be fastis_enabled=awaitclient.check_flag(
"feature-flag",
company={"id": "company-123"}
)
return {"feature_enabled": is_enabled}

Exception Handling

All errors thrown by the SDK will be subclasses of ApiError.

try:
client.companies.get_company(
company_id="company_id",
)
exceptschematic.core.ApiErrorase: # Handle all errorsprint(e.status_code)
print(e.body)

Usage examples

A number of these examples use keys to identify companies and users. Learn more about keys here.

Sending identify events

Create or update users and companies using identify events.

fromschematicimportEventBodyIdentifyCompanyfromschematic.clientimportSchematicclient=Schematic("YOUR_API_KEY")
client.identify(
keys={
"email": "wcoyote@acme.net",
"user_id": "your-user-id",
},
company=EventBodyIdentifyCompany(
keys={"id": "your-company-id"},
name="Acme Widgets, Inc.",
traits={
"city": "Atlanta",
},
),
name="Wile E. Coyote",
traits={
"login_count": 24,
"is_staff": false,
},
)

This call is non-blocking and there is no response to check.

Sending track events

Track activity in your application using track events; these events can later be used to produce metrics for targeting.

fromschematic.clientimportSchematicclient=Schematic("YOUR_API_KEY")
client.track(
event="some-action",
user={"user_id": "your-user-id"},
company={"id": "your-company-id"},
)

Async client:

importasynciofromschematic.clientimportAsyncSchematicasyncdefmain():
asyncwithAsyncSchematic("YOUR_API_KEY") asclient:
awaitclient.track(
event="some-action",
user={"user_id": "your-user-id"},
company={"id": "your-company-id"},
)
asyncio.run(main())

These calls are non-blocking and there is no response to check.

If you want to record large numbers of the same event at once, or perhaps measure usage in terms of a unit like tokens or memory, you can optionally specify a quantity for your event:

client.track(
event="some-action",
user={"user_id": "your-user-id"},
company={"id": "your-company-id"},
quantity=10,
)

Creating and updating companies

Although it is faster to create companies and users via identify events, if you need to handle a response, you can use the companies API to upsert companies. Because you use your own identifiers to identify companies, rather than a Schematic company ID, creating and updating companies are both done via the same upsert operation:

fromschematic.clientimportSchematicclient=Schematic("YOUR_API_KEY")
client.companies.upsert_company(
keys={"id": "your-company-id"},
name="Acme Widgets, Inc.",
traits={
"city": "Atlanta",
"high_score": 25,
"is_active": true,
},
)

You can define any number of company keys; these are used to address the company in the future, for example by updating the company's traits or checking a flag for the company.

You can also define any number of company traits; these can then be used as targeting parameters.

Creating and updating users

Similarly, you can upsert users using the Schematic API, as an alternative to using identify events. Because you use your own identifiers to identify users, rather than a Schematic user ID, creating and updating users are both done via the same upsert operation:

fromschematic.clientimportSchematicclient=Schematic("YOUR_API_KEY")
client.companies.upsert_user(
keys={
"email": "wcoyote@acme.net",
"user_id": "your-user-id",
},
name="Wile E. Coyote",
traits={
"city": "Atlanta",
"high_score": 25,
"is_active": true,
},
company={"id": "your-company-id"},
)

You can define any number of user keys; these are used to address the user in the future, for example by updating the user's traits or checking a flag for the user.

You can also define any number of user traits; these can then be used as targeting parameters.

Checking flags

When checking a flag, you'll provide keys for a company and/or keys for a user. You can also provide no keys at all, in which case you'll get the default value for the flag.

fromschematic.clientimportSchematicclient=Schematic("YOUR_API_KEY")
client.check_flag(
"some-flag-key",
company={"id": "your-company-id"},
user={"user_id": "your-user-id"},
)

Webhook Verification

Schematic can send webhooks to notify your application of events. To ensure the security of these webhooks, Schematic signs each request using HMAC-SHA256. The Python SDK provides utility functions to verify these signatures.

Verifying Webhook Signatures

When your application receives a webhook request from Schematic, you should verify its signature to ensure it's authentic. The SDK provides simple functions to verify webhook signatures. Here's how to use them in different frameworks:

Flask

fromflaskimportFlask, request, jsonifyfromschematic.webhook_utilsimportverify_webhook_signature, WebhookSignatureErrorapp=Flask(__name__)
@app.route('/webhooks/schematic', methods=['POST'])defschematic_webhook():
try:
# Each webhook has a distinct secret; you can access this via the Schematic appwebhook_secret="your-webhook-secret"# Verify the webhook signatureverify_webhook_signature(request, webhook_secret)
# Process the webhook payloaddata=request.jsonprint(f"Webhook verified: {data}")
return"", 200exceptWebhookSignatureErrorase:
print(f"Webhook verification failed: {str(e)}")
returnjsonify({"error": str(e)}), 400exceptExceptionase:
print(f"Error processing webhook: {str(e)}")
returnjsonify({"error": "Internal server error"}), 500if__name__=='__main__':
app.run(port=3000)

Django

fromdjango.httpimportJsonResponse, HttpResponsefromdjango.views.decorators.csrfimportcsrf_exemptfromschematic.webhook_utilsimportverify_webhook_signature, WebhookSignatureError@csrf_exemptdefschematic_webhook(request):
ifrequest.method!='POST':
returnHttpResponse(status=405)
try:
# Each webhook has a distinct secret; you can access this via the Schematic appwebhook_secret="your-webhook-secret"# Verify the webhook signatureverify_webhook_signature(request, webhook_secret)
# Process the webhook payloaddata=request.jsonprint(f"Webhook verified: {data}")
returnHttpResponse(status=200)
exceptWebhookSignatureErrorase:
print(f"Webhook verification failed: {str(e)}")
returnJsonResponse({"error": str(e)}, status=400)
exceptExceptionase:
print(f"Error processing webhook: {str(e)}")
returnJsonResponse({"error": "Internal server error"}, status=500)

FastAPI

fromfastapiimportFastAPI, Request, Response, HTTPException, Dependsfromschematic.webhook_utilsimportverify_webhook_signature, WebhookSignatureErrorapp=FastAPI()
asyncdefverify_signature(request: Request):
# Each webhook has a distinct secret; you can access this via the Schematic appwebhook_secret="your-webhook-secret"try:
# Get the raw bodybody=awaitrequest.body()
# Verify the webhook signatureverify_webhook_signature(request, webhook_secret, body)
exceptWebhookSignatureErrorase:
raiseHTTPException(status_code=400, detail=str(e))
@app.post("/webhooks/schematic")asyncdefschematic_webhook(request: Request, _: None=Depends(verify_signature)):
# Process the webhook payloaddata=awaitrequest.json()
print(f"Webhook verified: {data}")
returnResponse(status_code=200)

Verifying Signatures Manually

If you need to verify a webhook signature outside of the context of a web request, you can use the verify_signature function:

fromschematic.webhook_utilsimportverify_signature, WebhookSignatureErrordefverify_webhook_manually(body: str, signature: str, timestamp: str, secret: str):
try:
# Verify the signatureverify_signature(body, signature, timestamp, secret)
returnTrueexceptWebhookSignatureErrorase:
print(f"Webhook verification failed: {str(e)}")
returnFalse

Advanced

Flag Check Options

By default, the client will do some local caching for flag checks. If you would like to change this behavior, you can do so using an initialization option to specify the max size of the cache (in terms of number of entries) and the max age of the cache (in milliseconds):

fromschematic.clientimportLocalCache, Schematiccache_size=100cache_ttl=1000# in millisecondsconfig=SchematicConfig(
cache_providers=[LocalCache[bool](cache_size, cache_ttl)],
)
client=Schematic("YOUR_API_KEY", config)

You can also disable local caching entirely; bear in mind that, in this case, every flag check will result in a network request:

fromschematic.clientimportSchematicconfig=SchematicConfig(cache_providers=[])
client=Schematic("YOUR_API_KEY", config)

You may want to specify default flag values for your application, which will be used if there is a service interruption or if the client is running in offline mode (see below):

fromschematic.clientimportSchematicconfig=SchematicConfig(flag_defaults={"some-flag-key": True})
client=Schematic("YOUR_API_KEY", config)

Offline Mode

In development or testing environments, you may want to avoid making network requests to the Schematic API. You can run Schematic in offline mode by specifying the offline option; in this case, it does not matter what API key you specify:

fromschematic.clientimportSchematicconfig=SchematicConfig(offline=True)
client=Schematic("", config)

Offline mode works well with flag defaults:

fromschematic.clientimportSchematicconfig=SchematicConfig(
flag_defaults={"some-flag-key": True},
offline=True,
)
client=Schematic("", config)
client.check_flag("some-flag-key") # Returns True

You can also set flag defaults dynamically after the client has been constructed using set_flag_default and set_flag_defaults. This is useful in automated testing contexts, where you may want to specify per-test flag values:

fromschematic.clientimportSchematic, SchematicConfigclient=Schematic("", SchematicConfig(offline=True))
# Set a single flag defaultclient.set_flag_default("some-flag-key", True)
# Or set multiple flag defaults at onceclient.set_flag_defaults({
"some-flag-key": True,
"another-flag-key": False,
})
client.check_flag("some-flag-key") # Returns True

Timeouts

By default, requests time out after 60 seconds. You can configure this with a timeout option at the client or request level.

fromschematic.clientimportSchematicclient=Schematic(
# All timeouts are 20 secondstimeout=20.0,
)
# Override timeout for a specific methodclient.companies.get_company(..., {
timeout_in_seconds=20.0
})

Retries

The SDK is instrumented with automatic retries with exponential backoff. A request will be retried as long as the request is deemed retriable and the number of retry attempts has not grown larger than the configured retry limit (default: 2).

A request is deemed retriable when any of the following HTTP status codes is returned:

• 408 (Timeout)
• 429 (Too Many Requests)
• 5XX (Internal Server Errors)

Use the max_retries request option to configure this behavior.

# Override timeout for a specific methodclient.companies.get_company(..., {
max_retries=1# Only retry once on failure
})

Custom HTTP client

You can override the httpx client to customize it for your use-case. Some common use-cases include support for proxies and transports.

importhttpxfromschematic.clientimportSchematicclient=Schematic(
http_client=httpx.Client(
proxies="http://my.test.proxy.example.com",
transport=httpx.HTTPTransport(local_address="0.0.0.0"),
),
)

DataStream

DataStream enables local flag evaluation by maintaining a WebSocket connection to Schematic and caching flag rules, company, and user data locally (or in a shared cache such as Redis). Flag checks are evaluated locally via a WASM rules engine, eliminating per-check network requests.

Async-only: DataStream and Replicator Mode are only available on the AsyncSchematic client. The synchronous Schematic client does not support either feature — use AsyncSchematic (shown in all examples below) if you need them.

Installation

DataStream requires additional dependencies for WebSocket connections and local flag evaluation. Install them with the datastream extra:

pip install 'schematichq[datastream]'# or
poetry add schematichq -E datastream

To use the Redis-backed shared cache (see below), also install redis:

pip install 'schematichq[datastream]' redis

Key Features

• Real-Time Updates: Automatically updates cached data when changes occur on the backend.
• Local Flag Evaluation: Flag checks are evaluated locally via WASM, eliminating per-check network requests.
• Configurable Caching: Supports in-memory caching (default) and custom AsyncCacheProvider implementations including a built-in Redis provider.

How to Enable DataStream

Set use_datastream=True on AsyncSchematicConfig:

importasynciofromschematic.clientimportAsyncSchematic, AsyncSchematicConfig, DataStreamConfigasyncdefmain():
config=AsyncSchematicConfig(
use_datastream=True,
datastream=DataStreamConfig(
cache_ttl=300_000, # 5 minutes, in ms
),
)
asyncwithAsyncSchematic("YOUR_API_KEY", config) asclient:
is_enabled=awaitclient.check_flag(
"some-flag-key",
company={"id": "your-company-id"},
user={"id": "your-user-id"},
)
asyncio.run(main())

Configuration Options

All fields live on DataStreamConfig.

Option	Type	Default	Description
cache_ttl	Optional[int]	24 hours	Cache TTL in milliseconds. None means no expiration.
company_cache	AsyncCacheProvider	in-memory	Cache for full company records.
company_lookup_cache	AsyncCacheProvider	in-memory	Cache mapping company keys → company IDs.
user_cache	AsyncCacheProvider	in-memory	Cache for full user records.
user_lookup_cache	AsyncCacheProvider	in-memory	Cache mapping user keys → user IDs.
flag_cache	AsyncCacheProvider	in-memory	Cache for flag rules.
replicator_mode	bool	False	Enable Replicator Mode (see below).
replicator_health_url	Optional[str]	http://localhost:8090/ready	Replicator health check URL.
replicator_health_check	Optional[int]	30000	Health check interval in milliseconds.

Using Redis as a Shared Cache

The SDK ships with a RedisCache provider built on redis.asyncio. Pass a Redis client into the cache slots on DataStreamConfig to share state across multiple processes:

importasyncioimportredis.asyncioasaioredisfromschematic.cacheimportRedisCachefromschematic.clientimportAsyncSchematic, AsyncSchematicConfig, DataStreamConfigasyncdefmain():
redis_client=aioredis.from_url("redis://localhost:6379")
cache_ttl_ms=60*60*1000# 1 hourconfig=AsyncSchematicConfig(
use_datastream=True,
datastream=DataStreamConfig(
cache_ttl=cache_ttl_ms,
company_cache=RedisCache(redis_client, default_ttl_ms=cache_ttl_ms),
company_lookup_cache=RedisCache(redis_client, default_ttl_ms=cache_ttl_ms),
user_cache=RedisCache(redis_client, default_ttl_ms=cache_ttl_ms),
user_lookup_cache=RedisCache(redis_client, default_ttl_ms=cache_ttl_ms),
flag_cache=RedisCache(redis_client, default_ttl_ms=cache_ttl_ms),
),
)
asyncwithAsyncSchematic("YOUR_API_KEY", config) asclient:
awaitclient.check_flag("some-flag-key", company={"id": "your-company-id"})
asyncio.run(main())

RedisCache accepts a prefix argument (default "schematic") if you need to namespace keys — this must match the prefix used by any other SDKs or the replicator writing to the same Redis instance.

Replicator Mode

When running the schematic-datastream-replicator service, configure the client to operate in Replicator Mode. The replicator holds the single WebSocket connection to Schematic and populates a shared cache; SDK instances read from that cache and evaluate flags locally without opening their own WebSocket connections.

Replicator Mode requires a shared cache (e.g. Redis) so the SDK can read data written by the external replicator process. Configure the cache slots on DataStreamConfig exactly as in the Redis example above.

How to Enable Replicator Mode

importasyncioimportredis.asyncioasaioredisfromschematic.cacheimportRedisCachefromschematic.clientimportAsyncSchematic, AsyncSchematicConfig, DataStreamConfigasyncdefmain():
redis_client=aioredis.from_url("redis://localhost:6379")
config=AsyncSchematicConfig(
use_datastream=True,
datastream=DataStreamConfig(
replicator_mode=True,
cache_ttl=None, # Match the replicator's unlimited defaultcompany_cache=RedisCache(redis_client),
company_lookup_cache=RedisCache(redis_client),
user_cache=RedisCache(redis_client),
user_lookup_cache=RedisCache(redis_client),
flag_cache=RedisCache(redis_client),
),
)
asyncwithAsyncSchematic("YOUR_API_KEY", config) asclient:
is_enabled=awaitclient.check_flag(
"some-flag-key",
company={"id": "your-company-id"},
)
asyncio.run(main())

Cache TTL Configuration

Set the SDK's cache_ttl to match the replicator's cache TTL. The replicator defaults to an unlimited cache TTL. If the SDK uses a shorter TTL (the default is 24 hours), locally updated cache entries (e.g. after track events) will be written back with the shorter TTL and eventually evicted from the shared cache, even though the replicator originally set them with no expiration.

If you have configured a custom cache TTL on the replicator, use the same value here.

Advanced Configuration

The client automatically configures sensible defaults for Replicator Mode, but you can customize the health check endpoint and interval:

config=AsyncSchematicConfig(
use_datastream=True,
datastream=DataStreamConfig(
replicator_mode=True,
cache_ttl=None,
replicator_health_url="http://my-replicator:8090/ready",
replicator_health_check=60_000, # 60 seconds, in ms# ... shared cache providers
),
)

Default Configuration

• Replicator Health URL: http://localhost:8090/ready
• Health Check Interval: 30 seconds
• Cache TTL: 24 hours (SDK default; should be set to match the replicator's TTL, which defaults to unlimited)

When running in Replicator Mode, the client will:

• Skip establishing WebSocket connections
• Periodically check if the replicator service is ready
• Use cached data populated by the external replicator service
• Fall back to direct API calls if the replicator is not available

Contributing

While we value open-source contributions to this SDK, this library is generated programmatically. Additions made directly to this library would have to be moved over to our generation code, otherwise they would be overwritten upon the next generated release. Feel free to open a PR as a proof of concept, but know that we will not be able to merge it as-is. We suggest opening an issue first to discuss with us!

On the other hand, contributions to the README are always very welcome!