Getting started¶

# MongoDB for the examples: a single-node replica set on localhost:27017.
# Transactions and change streams need a replica set; one node is enough.
#   docker compose -f examples/compose.yaml up -d --wait
name: mongospecs-examples
services:
  mongo:
    image: mongo:8.0
    command: ["mongod", "--replSet", "rs0", "--bind_ip_all"]
    ports:
      - "27017:27017"
    volumes:
      - data:/data/db
    healthcheck:
      test:
        - CMD-SHELL
        - >
          mongosh --quiet --eval
          "try { if (!db.hello().isWritablePrimary) throw 1 } catch (e) {
          try { rs.initiate({_id: 'rs0', members: [{_id: 0, host: 'localhost:27017'}]}) } catch (e2) {}
          quit(1) }"
      interval: 2s
      timeout: 10s
      retries: 60
volumes:
  data:

docker compose -f compose.yaml up -d --wait

pip install "mongospecs[msgspec]"        # or [pydantic], or [attrs]

"""mongospecs in one file. Start MongoDB (see compose.yaml), then: python examples/quickstart.py"""

import asyncio
import os
from datetime import datetime
from typing import Annotated

from mongospecs import CreatedAt, Inc, Index, Q, Ref, Registry, Revision, SpecConfig, UpdatedAt
from mongospecs.msgspec import Spec  # or mongospecs.pydantic / mongospecs.attrs

URI = os.environ.get("MONGO_URI", "mongodb://localhost:27017/quickstart?replicaSet=rs0")


class User(Spec):
    name: str = ""


class Article(Spec):
    __spec__ = SpecConfig(collection="articles", indexes=[Index("title", unique=True)])

    title: str
    views: int = 0
    tags: list[str] = []
    author: Ref[User] | None = None
    created_at: Annotated[datetime | None, CreatedAt] = None
    updated_at: Annotated[datetime | None, UpdatedAt] = None
    rev: Annotated[int, Revision] = 0  # optimistic locking


async def main(uri: str = URI) -> None:
    async with Registry(uri, models=[User, Article]) as registry:
        await registry.apply_indexes()

        ann = await User(name="ann").insert()
        article = await Article(title="Hello", author=ann).insert()

        # change tracking: save() sends only what changed
        article.title = "Hello, world"
        assert article.changes() == {"$set": {"title": "Hello, world"}}
        await article.save()

        # typed queries: names are checked, values are cast
        popular = await Article.find(Q.views >= 0, Q.title.prefix("Hello")).sort("-views").limit(10)
        assert [a.title for a in popular] == ["Hello, world"]

        # relations: one extra query, or a $lookup
        with_author = await Article.find(Q.author == ann).include("author").one()
        print(with_author.title, "by", with_author.author.doc.name)

        # atomic updates, always validated; the updated document comes back
        article = await Article.update(article.id, Inc(views=1), tags=["intro"])
        assert article.views == 1

        # transactions (these need a replica set): the callback is retried on transient
        # errors, and in-memory documents are restored if it aborts
        async def publish(tx):
            article.tags.append("published")
            await article.save(session=tx)

        await registry.transaction(publish)
        print("tags:", (await Article.get(article.id)).tags)


if __name__ == "__main__":
    asyncio.run(main())

$ python quickstart.py
Hello, world by ann
tags: ['intro', 'published']

pip install mongospecs-web uvicorn

"""A small REST API with mongospecs-web: per-user to-do lists.

    pip install "mongospecs[msgspec]" mongospecs-web uvicorn
    docker compose -f examples/compose.yaml up -d --wait
    uvicorn app:app --app-dir examples/todo_api

    curl -s -X POST localhost:8000/api/tasks -H 'authorization: Bearer ann-token' \
         -H 'content-type: application/json' -d '{"title": "Write docs"}'
    curl -s 'localhost:8000/api/tasks?done=false&sort=-title' -H 'authorization: Bearer ann-token'
"""

import os
from datetime import datetime
from typing import Annotated

from mongospecs import CreatedAt, Index, Policy, Q, Registry, Revision, SpecConfig
from mongospecs.msgspec import Spec
from mongospecs_web import Api, Create, Delete, Get, List, Patch, Resource, Unauthorized

URI = os.environ.get("MONGO_URI", "mongodb://localhost:27017/todo?replicaSet=rs0")

# Stands in for your real user store or token verification.
TOKENS = {"ann-token": "ann", "bob-token": "bob"}


class Task(Spec, kw_only=True):
    __spec__ = SpecConfig(collection="tasks", indexes=[Index("owner", "title")])

    # Policy says what a client may do with each field.
    title: Annotated[str, Policy(filter=("eq", "prefix"), sort=True)]
    done: Annotated[bool, Policy(filter=("eq",))] = False
    owner: Annotated[str, Policy(insert=False, update=False)] = ""  # set by the server
    note_to_self: Annotated[str | None, Policy(private=True)] = None  # never leaves the server
    created_at: Annotated[datetime | None, CreatedAt] = None
    rev: Annotated[int, Revision] = 0  # the ETag, checked by If-Match


def authenticate(request):
    """Who is calling: the principal, or None for an anonymous request."""
    scheme, _, token = request.headers.get("authorization", "").partition(" ")
    if not token:
        return None
    if scheme != "Bearer" or token not in TOKENS:
        raise Unauthorized("Unknown token.", headers={"www-authenticate": "Bearer"})
    return TOKENS[token]


def logged_in(ctx):
    if ctx.principal is None:
        raise Unauthorized("Log in first.", headers={"www-authenticate": "Bearer"})
    return True


tasks = Resource(
    Task,
    ops=[List, Get, Create, Patch, Delete],
    guards=[logged_in],  # who may call the operations
    scope=lambda ctx: Q.owner == ctx.principal,  # which rows they may see and change
    server_fields=lambda ctx: {"owner": ctx.principal},  # values the server sets on create
)


@tasks.op("complete", method="POST", summary="Mark a task as done")
async def complete(ctx):
    task = await ctx.load()  # 404 unless the task is within the caller's scope
    task.done = True
    return await task.save()


registry = Registry(URI, models=[Task])
api = Api(tasks, prefix="/api", title="To-do", authenticate=authenticate)
app = api.asgi(registry=registry)  # opens the registry on startup and closes it on shutdown

$ uvicorn app:app
$ curl -s -X POST localhost:8000/api/tasks -H 'authorization: Bearer ann-token' \
       -H 'content-type: application/json' -d '{"title": "Write docs"}'
{"data":{"id":"...","title":"Write docs","done":false,"owner":"ann","created_at":"...","rev":0}}
$ curl -s 'localhost:8000/api/tasks?done=false&sort=-title' -H 'authorization: Bearer ann-token'
$ curl -s localhost:8000/api/openapi.json

"""Testing the API without a server: api.handle() runs a request through the whole stack."""

import json

import pytest_asyncio

from app import URI, Task, api
from mongospecs import Registry

ANN = {"authorization": "Bearer ann-token", "content-type": "application/json"}
BOB = {"authorization": "Bearer bob-token", "content-type": "application/json"}


@pytest_asyncio.fixture(autouse=True)
async def database():
    # A registry cannot be reopened once closed, so each test opens its own.
    async with Registry(URI, models=[Task]) as registry:
        yield
        await registry.client.drop_database(registry.database().name)


async def call(method, path, body=None, headers=ANN):
    response = await api.handle(
        method, path, body=json.dumps(body).encode() if body is not None else b"", headers=headers
    )
    return response.status, json.loads(response.body) if response.body else None, response.headers


async def test_create_list_and_complete():
    status, created, headers = await call("POST", "/api/tasks", {"title": "Write docs"})
    assert status == 201 and created["data"]["owner"] == "ann" and headers["etag"] == '"0"'
    assert "note_to_self" not in created["data"]  # private fields are never returned

    status, listed, _ = await call("GET", "/api/tasks")
    assert [task["title"] for task in listed["data"]] == ["Write docs"]

    status, done, _ = await call("POST", f"{headers['location']}/complete")
    assert status == 200 and done["data"]["done"] is True


async def test_each_user_sees_only_their_own_tasks():
    _, _, headers = await call("POST", "/api/tasks", {"title": "Ann's"})
    assert (await call("GET", headers["location"], headers=BOB))[0] == 404
    assert (await call("GET", "/api/tasks", headers=BOB))[1]["data"] == []
    assert (await call("GET", "/api/tasks", headers={}))[0] == 401


async def test_the_field_policy_is_enforced():
    _, _, headers = await call("POST", "/api/tasks", {"title": "Mine"})
    status, problem, _ = await call("PATCH", headers["location"], {"owner": "bob"})
    assert status == 422 and problem["errors"] == [{"field": "owner", "message": "cannot be set on update"}]
    assert (await call("PATCH", headers["location"], {"note_to_self": "x"}))[0] == 422


async def test_a_stale_etag_is_refused():
    _, _, headers = await call("POST", "/api/tasks", {"title": "Mine"})
    fresh = {**ANN, "if-match": headers["etag"]}
    assert (await call("PATCH", headers["location"], {"done": True}, fresh))[0] == 200
    assert (await call("PATCH", headers["location"], {"done": False}, fresh))[0] == 412