Python

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

1
from schematic.client import AsyncSchematic
2
3
client = AsyncSchematic("YOUR_API_KEY")
4
async def main() -> None:
5
    await client.companies.get_company(
6
        company_id="company_id",
7
    )
8
9
asyncio.run(main())

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
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
)