Skip to content
Aerostack v1.0-beta

Python SDK

What you can build

Section titled “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

Section titled “Install”
Terminal window
pip install aerostack

Initialize

Section titled “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

Section titled “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

Section titled “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

Section titled “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

Section titled “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

Section titled “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

Section titled “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

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

Secrets

Section titled “Secrets”
stripe_key = await client.secrets.get("STRIPE_SECRET_KEY")

Complete example: FastAPI integration

Section titled “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

Section titled “Next steps”