SDK (Beta)PythonSetup & Guide

Python SDK

The Python SDK is generated from the OpenAPI spec. All endpoints are available. The API is async-first using asyncio.

⚠️

Beta — The Python SDK is in active beta. APIs may change between minor versions.

What you can build

  • Django/FastAPI backends — Add Aerostack services to existing Python web apps
  • Data pipelines — ETL scripts that read/write to your Aerostack database
  • ML workflows — Store embeddings, run semantic search, use AI completions
  • Admin scripts — Bulk user management, data migration, reporting
  • Jupyter notebooks — Interactive data exploration and prototyping
  • Discord/Slack bots — Python-based bots using Aerostack auth and database

Install

pip install aerostack

Initialize

from aerostack import AerostackClient
 
client = AerostackClient(
    project_id="your-project-id",
    api_key="your-api-key",
    base_url="https://api.aerostack.dev/v1"
)

Auth

# Register a user
result = await client.auth.register(
    email="jane@example.com",
    password="securePassword123",
    name="Jane Doe"
)
 
# Login
result = await client.auth.login(
    email="jane@example.com",
    password="securePassword123"
)
access_token = result.access_token
 
# Verify a token (server-side)
user = await client.auth.verify_token(token)
if not user:
    raise Exception("Unauthorized")
 
# Passwordless OTP
await client.auth.send_otp(email="jane@example.com")
result = await client.auth.verify_otp(code="482916", email="jane@example.com")

Database

# Query multiple rows
users = await client.db.query("SELECT * FROM users WHERE active = 1")
 
# Query single row
user = await client.db.query_one(
    "SELECT * FROM users WHERE id = ?",
    [user_id]
)
 
# Insert
await client.db.query(
    "INSERT INTO users (id, email, name) VALUES (?, ?, ?)",
    [user_id, email, name]
)
 
# Batch (atomic transaction)
await client.db.batch([
    {"sql": "INSERT INTO orders ...", "params": [...]},
    {"sql": "UPDATE inventory ...", "params": [...]},
])

Cache

# Set with TTL (seconds)
await client.cache.set("user:123", user_data, ttl=300)
 
# Get (None if not found)
user = await client.cache.get("user:123")
 
# Check existence
exists = await client.cache.has("user:123")
 
# Delete
await client.cache.delete("user:123")

Storage

result = await client.storage.upload(
    file_bytes=image_data,
    path="avatars/user-123.jpg",
    content_type="image/jpeg"
)
print(result.url)  # CDN URL

AI

# Chat completion
result = await client.ai.complete(
    prompt="Summarize this document in 3 bullet points.",
    model="gpt-4o-mini",
    max_tokens=256,
    temperature=0.7
)
print(result.text)
 
# Embeddings
vector = await client.ai.embed("text to embed")

Realtime

channel = client.realtime.channel("orders")
 
@channel.on("INSERT")
async def on_insert(payload):
    print(f"New order: {payload.data}")
 
@channel.on("UPDATE")
async def on_update(payload):
    print(f"Order updated: {payload.data}")
 
channel.subscribe()
 
# Publish custom events
channel.publish("notification", {"text": "New order received"})
 
# Presence tracking
channel.track({"userId": "123", "name": "Bot", "status": "online"})
 
# Server-side broadcast
client.socket.emit("order:updated", {"id": order_id}, f"orders/{order_id}")
 
# Message history
messages = await channel.get_history(limit=50)

Queue

await client.queue.send(
    "send-welcome-email",
    {"userId": "user-123", "email": "jane@example.com"},
    delay_seconds=0,
    max_retries=3
)

Secrets

stripe_key = await client.secrets.get("STRIPE_SECRET_KEY")

Complete example: FastAPI integration

main.py
from fastapi import FastAPI, Header, HTTPException
from aerostack import AerostackClient
 
app = FastAPI()
client = AerostackClient(
    project_id="your-project-id",
    api_key="your-api-key"
)
 
@app.get("/api/users")
async def list_users(authorization: str = Header()):
    # Verify auth
    token = authorization.replace("Bearer ", "")
    user = await client.auth.verify_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Unauthorized")
 
    # Query database
    users = await client.db.query(
        "SELECT id, name, email FROM users ORDER BY created_at DESC LIMIT 50"
    )
    return users
 
@app.post("/api/tasks")
async def create_task(body: dict, authorization: str = Header()):
    token = authorization.replace("Bearer ", "")
    user = await client.auth.verify_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Unauthorized")
 
    result = await client.db.query(
        "INSERT INTO tasks (id, user_id, title, status) VALUES (?, ?, ?, ?) RETURNING *",
        [generate_id(), user["id"], body["title"], "todo"]
    )
    return result[0]

Next steps

API ReferenceAPI Reference