Schematic Python Library

1
from schematic.client import Schematic
2
3
client = Schematic("YOUR_API_KEY")

1
import asyncio
2
from schematic.client import AsyncSchematic
3
4
async def main():
5
    # Create client - no initialize() needed!
6
    client = AsyncSchematic("YOUR_API_KEY")
7
    
8
    # Use immediately - auto-initializes on first call
9
    is_enabled = await client.check_flag(
10
        "new-feature",
11
        company={"id": "company-123"},
12
        user={"id": "user-456"}
13
    )
14
    
15
    if is_enabled:
16
        print("New feature is enabled!")
17
    
18
    # Track usage
19
    await client.track(
20
        event="feature-used",
21
        company={"id": "company-123"},
22
        user={"id": "user-456"}
23
    )
24
    
25
    # Always shutdown when done
26
    await client.shutdown()
27
28
asyncio.run(main())

1
import asyncio
2
from schematic.client import AsyncSchematic
3
4
async def main():
5
    async with AsyncSchematic("YOUR_API_KEY") as client:
6
        # Client auto-initializes and will auto-shutdown
7
        
8
        is_enabled = await client.check_flag(
9
            "feature-flag",
10
            company={"id": "company-123"}
11
        )
12
        
13
        await client.identify(
14
            keys={"id": "company-123"},
15
            name="Acme Corp"
16
        )
17
        
18
        # Automatic cleanup on exit
19
20
asyncio.run(main())

1
import asyncio
2
from schematic.client import AsyncSchematic
3
4
# Web application example
5
client = AsyncSchematic("YOUR_API_KEY")
6
7
async def startup():
8
    """Call during application startup"""
9
    await client.initialize()  # Start background tasks now
10
    print("Schematic client ready")
11
12
async def shutdown():
13
    """Call during application shutdown"""  
14
    await client.shutdown()   # Stop background tasks and flush events
15
    print("Schematic client stopped")
16
17
async def handle_request():
18
    """Handle individual requests"""
19
    # Client is already initialized - this will be fast
20
    is_enabled = await client.check_flag(
21
        "feature-flag",
22
        company={"id": "company-123"}
23
    )
24
    return {"feature_enabled": is_enabled}

1
try:
2
    client.companies.get_company(
3
        company_id="company_id",
4
    )
5
except schematic.core.ApiError as e: # Handle all errors
6
  print(e.status_code)
7
  print(e.body)

1
from schematic import EventBodyIdentifyCompany
2
from schematic.client import Schematic
3
4
client = Schematic("YOUR_API_KEY")
5
6
client.identify(
7
    keys={
8
        "email": "wcoyote@acme.net",
9
        "user_id": "your-user-id",
10
    },
11
    company=EventBodyIdentifyCompany(
12
        keys={"id": "your-company-id"},
13
        name="Acme Widgets, Inc.",
14
        traits={
15
          "city": "Atlanta",
16
        },
17
    ),
18
    name="Wile E. Coyote",
19
    traits={
20
        "login_count": 24,
21
        "is_staff": false,
22
    },
23
)

1
from schematic.client import Schematic
2
3
client = Schematic("YOUR_API_KEY")
4
5
client.track(
6
    event="some-action",
7
    user={"user_id": "your-user-id"},
8
    company={"id": "your-company-id"},
9
)

1
import asyncio
2
from schematic.client import AsyncSchematic
3
4
async def main():
5
    async with AsyncSchematic("YOUR_API_KEY") as client:
6
        await client.track(
7
            event="some-action",
8
            user={"user_id": "your-user-id"},
9
            company={"id": "your-company-id"},
10
        )
11
12
asyncio.run(main())

1
client.track(
2
    event="some-action",
3
    user={"user_id": "your-user-id"},
4
    company={"id": "your-company-id"},
5
    quantity=10,
6
)

1
from schematic.client import Schematic
2
3
client = Schematic("YOUR_API_KEY")
4
5
client.companies.upsert_company(
6
    keys={"id": "your-company-id"},
7
    name="Acme Widgets, Inc.",
8
    traits={
9
        "city": "Atlanta",
10
        "high_score": 25,
11
        "is_active": true,
12
    },
13
)

1
from schematic.client import Schematic
2
3
client = Schematic("YOUR_API_KEY")
4
5
client.companies.upsert_user(
6
    keys={
7
        "email": "wcoyote@acme.net",
8
        "user_id": "your-user-id",
9
    },
10
    name="Wile E. Coyote",
11
    traits={
12
        "city": "Atlanta",
13
        "high_score": 25,
14
        "is_active": true,
15
    },
16
    company={"id": "your-company-id"},
17
)

1
from schematic.client import Schematic
2
3
client = Schematic("YOUR_API_KEY")
4
5
client.check_flag(
6
    "some-flag-key",
7
    company={"id": "your-company-id"},
8
    user={"user_id": "your-user-id"},
9
)

1
from flask import Flask, request, jsonify
2
from schematic.webhook_utils import verify_webhook_signature, WebhookSignatureError
3
4
app = Flask(__name__)
5
6
@app.route('/webhooks/schematic', methods=['POST'])
7
def schematic_webhook():
8
    try:
9
        # Each webhook has a distinct secret; you can access this via the Schematic app
10
        webhook_secret = "your-webhook-secret"
11
12
        # Verify the webhook signature
13
        verify_webhook_signature(request, webhook_secret)
14
15
        # Process the webhook payload
16
        data = request.json
17
        print(f"Webhook verified: {data}")
18
19
        return "", 200
20
    except WebhookSignatureError as e:
21
        print(f"Webhook verification failed: {str(e)}")
22
        return jsonify({"error": str(e)}), 400
23
    except Exception as e:
24
        print(f"Error processing webhook: {str(e)}")
25
        return jsonify({"error": "Internal server error"}), 500
26
27
if __name__ == '__main__':
28
    app.run(port=3000)

1
from django.http import JsonResponse, HttpResponse
2
from django.views.decorators.csrf import csrf_exempt
3
from schematic.webhook_utils import verify_webhook_signature, WebhookSignatureError
4
5
@csrf_exempt
6
def schematic_webhook(request):
7
    if request.method != 'POST':
8
        return HttpResponse(status=405)
9
10
    try:
11
        # Each webhook has a distinct secret; you can access this via the Schematic app
12
        webhook_secret = "your-webhook-secret"
13
14
        # Verify the webhook signature
15
        verify_webhook_signature(request, webhook_secret)
16
17
        # Process the webhook payload
18
        data = request.json
19
        print(f"Webhook verified: {data}")
20
21
        return HttpResponse(status=200)
22
    except WebhookSignatureError as e:
23
        print(f"Webhook verification failed: {str(e)}")
24
        return JsonResponse({"error": str(e)}, status=400)
25
    except Exception as e:
26
        print(f"Error processing webhook: {str(e)}")
27
        return JsonResponse({"error": "Internal server error"}, status=500)

1
from fastapi import FastAPI, Request, Response, HTTPException, Depends
2
from schematic.webhook_utils import verify_webhook_signature, WebhookSignatureError
3
4
app = FastAPI()
5
6
async def verify_signature(request: Request):
7
    # Each webhook has a distinct secret; you can access this via the Schematic app
8
    webhook_secret = "your-webhook-secret"
9
10
    try:
11
        # Get the raw body
12
        body = await request.body()
13
14
        # Verify the webhook signature
15
        verify_webhook_signature(request, webhook_secret, body)
16
    except WebhookSignatureError as e:
17
        raise HTTPException(status_code=400, detail=str(e))
18
19
@app.post("/webhooks/schematic")
20
async def schematic_webhook(request: Request, _: None = Depends(verify_signature)):
21
    # Process the webhook payload
22
    data = await request.json()
23
    print(f"Webhook verified: {data}")
24
25
    return Response(status_code=200)

1
from schematic.webhook_utils import verify_signature, WebhookSignatureError
2
3
def verify_webhook_manually(body: str, signature: str, timestamp: str, secret: str):
4
    try:
5
        # Verify the signature
6
        verify_signature(body, signature, timestamp, secret)
7
        return True
8
    except WebhookSignatureError as e:
9
        print(f"Webhook verification failed: {str(e)}")
10
        return False

1
from schematic.client import LocalCache, Schematic
2
3
cache_size = 100
4
cache_ttl = 1000  # in milliseconds
5
config = SchematicConfig(
6
    cache_providers=[LocalCache[bool](cache_size, cache_ttl)],
7
)
8
client = Schematic("YOUR_API_KEY", config)

1
from schematic.client import Schematic
2
3
config = SchematicConfig(cache_providers=[])
4
client = Schematic("YOUR_API_KEY", config)

1
from schematic.client import Schematic
2
3
config = SchematicConfig(flag_defaults={"some-flag-key": True})
4
client = Schematic("YOUR_API_KEY", config)

1
from schematic.client import Schematic
2
3
config = SchematicConfig(offline=True)
4
client = Schematic("", config)

1
from schematic.client import Schematic
2
3
config = SchematicConfig(
4
    flag_defaults={"some-flag-key": True},
5
    offline=True,
6
)
7
client = Schematic("", config)
8
client.check_flag("some-flag-key") # Returns True

1
from schematic.client import Schematic
2
3
client = Schematic(
4
    # All timeouts are 20 seconds
5
    timeout=20.0,
6
)
7
8
# Override timeout for a specific method
9
client.companies.get_company(..., {
10
    timeout_in_seconds=20.0
11
})

1
# Override timeout for a specific method
2
client.companies.get_company(..., {
3
    max_retries=1 # Only retry once on failure
4
})

1
import httpx
2
3
from schematic.client import Schematic
4
5
client = Schematic(
6
    http_client=httpx.Client(
7
        proxies="http://my.test.proxy.example.com",
8
        transport=httpx.HTTPTransport(local_address="0.0.0.0"),
9
    ),
10
)

1
import asyncio
2
from schematic.client import AsyncSchematic, AsyncSchematicConfig, DataStreamConfig
3
4
async def main():
5
    config = AsyncSchematicConfig(
6
        use_datastream=True,
7
        datastream=DataStreamConfig(
8
            cache_ttl=300_000,  # 5 minutes, in ms
9
        ),
10
    )
11
12
    async with AsyncSchematic("YOUR_API_KEY", config) as client:
13
        is_enabled = await client.check_flag(
14
            "some-flag-key",
15
            company={"id": "your-company-id"},
16
            user={"id": "your-user-id"},
17
        )
18
19
asyncio.run(main())

1
import asyncio
2
import redis.asyncio as aioredis
3
from schematic.cache import RedisCache
4
from schematic.client import AsyncSchematic, AsyncSchematicConfig, DataStreamConfig
5
6
async def main():
7
    redis_client = aioredis.from_url("redis://localhost:6379")
8
    cache_ttl_ms = 60 * 60 * 1000  # 1 hour
9
10
    config = AsyncSchematicConfig(
11
        use_datastream=True,
12
        datastream=DataStreamConfig(
13
            cache_ttl=cache_ttl_ms,
14
            company_cache=RedisCache(redis_client, default_ttl_ms=cache_ttl_ms),
15
            company_lookup_cache=RedisCache(redis_client, default_ttl_ms=cache_ttl_ms),
16
            user_cache=RedisCache(redis_client, default_ttl_ms=cache_ttl_ms),
17
            user_lookup_cache=RedisCache(redis_client, default_ttl_ms=cache_ttl_ms),
18
            flag_cache=RedisCache(redis_client, default_ttl_ms=cache_ttl_ms),
19
        ),
20
    )
21
22
    async with AsyncSchematic("YOUR_API_KEY", config) as client:
23
        await client.check_flag("some-flag-key", company={"id": "your-company-id"})
24
25
asyncio.run(main())

1
import asyncio
2
import redis.asyncio as aioredis
3
from schematic.cache import RedisCache
4
from schematic.client import AsyncSchematic, AsyncSchematicConfig, DataStreamConfig
5
6
async def main():
7
    redis_client = aioredis.from_url("redis://localhost:6379")
8
9
    config = AsyncSchematicConfig(
10
        use_datastream=True,
11
        datastream=DataStreamConfig(
12
            replicator_mode=True,
13
            cache_ttl=None,  # Match the replicator's unlimited default
14
            company_cache=RedisCache(redis_client),
15
            company_lookup_cache=RedisCache(redis_client),
16
            user_cache=RedisCache(redis_client),
17
            user_lookup_cache=RedisCache(redis_client),
18
            flag_cache=RedisCache(redis_client),
19
        ),
20
    )
21
22
    async with AsyncSchematic("YOUR_API_KEY", config) as client:
23
        is_enabled = await client.check_flag(
24
            "some-flag-key",
25
            company={"id": "your-company-id"},
26
        )
27
28
asyncio.run(main())

1
config = AsyncSchematicConfig(
2
    use_datastream=True,
3
    datastream=DataStreamConfig(
4
        replicator_mode=True,
5
        cache_ttl=None,
6
        replicator_health_url="http://my-replicator:8090/ready",
7
        replicator_health_check=60_000,  # 60 seconds, in ms
8
        # ... shared cache providers
9
    ),
10
)