Infrastructure SDK for building
distributed systems in Python

from encore.storage import postgres

db = postgres.Database("users", migrations="./migrations")


async def get_user(user_id: int):
    return await db.query(
        "SELECT * FROM users WHERE id = $1", user_id
    ).one()

from encore import pubsub
from pydantic import BaseModel


class SignupEvent(BaseModel):
    user_id: str
    email: str


signups = pubsub.Topic("signups")


async def send_welcome(msg: pubsub.Message[SignupEvent]) -> None:
    await send_email(msg.data.email)


welcome = pubsub.Subscription[SignupEvent](
    signups, "send-welcome", handler=send_welcome
)

from encore.storage import objects

avatars = objects.Bucket("avatars", public=True)


async def save_avatar(user_id: str, image: bytes) -> str:
    key = f"{user_id}.png"
    await avatars.upload(key, image, content_type="image/png")
    return avatars.public_url(key)

from encore import cron

# Encore calls POST /internal/welcome-emails every 2 hours.
_ = cron.Job(
    "welcome-email",
    title="Send welcome emails",
    path="/internal/welcome-emails",
    every="2h",
)


@app.post("/internal/welcome-emails")
async def send_welcome_emails() -> None:
    ...

from encore.secrets import Secret

stripe_key = Secret[str]("StripeSecretKey")

# Read the value at runtime, no .env files to manage
client = StripeClient(stripe_key.value)

from encore.storage import redis

sessions = redis.Database("sessions")


async def track_visit(page: str) -> int:
    return await sessions.string.incr(f"visits:{page}")

from encore import metrics

orders_placed = metrics.Counter("orders_placed")
checkout_time = metrics.Histogram("checkout_duration", unit="seconds")

orders_placed.increment()
checkout_time.observe(0.42)