pypocketbase

An opinionated async PocketBase client for Python: aiohttp transport, pydantic models, and Result[T, PocketbaseException] returns instead of exceptions.

res = await pb.collection("posts").get_one(post_id)
if isinstance(res, Err):
    log.error(res.err_value.details)   # one rich diagnostic line
    return
post = res.ok_value                    # narrowed to Record

Nothing raises. A failed call hands back an Err carrying the status, the URL, the server's message and the arguments you passed — so the error path is something you handle, not something you remember to catch.

• Python 3.11+
• Covers records, auth, collections, batch, files, backups, logs, crons, settings and health
• Full type hints; isinstance narrowing works in Pyright and mypy

Install

uv add git+https://github.com/Touexe/pypocketbase
# or
pip install git+https://github.com/Touexe/pypocketbase

The distribution is pypocketbase; the import is pocketbase.

Quick start

import asyncio
from pocketbase import Pocketbase
from pocketbase.utils.params import ParamsList
from result import Err

async def main():
    async with Pocketbase(url="http://127.0.0.1:8090") as pb:
        auth = await pb.collection("users").auth_with_password("jane@example.com", "secret")
        if isinstance(auth, Err):
            print(auth.err_value.details)
            return

        res = await pb.collection("posts").list(
            ParamsList(per_page=20, sort="-created", filter='published = true', expand="author")
        )
        if isinstance(res, Err):
            print(res.err_value.details)
            return

        for post in res.ok_value.items:
            print(post.id, post.get("title"))

asyncio.run(main())

The client is an async context manager owning one aiohttp session; it closes on exit. auth_with_password writes the token into that session, so later calls on the same client are authenticated — which also means one client carries one identity. Serving several users concurrently wants a client per request, not a shared one.

Errors

Every service method returns a Result. Narrow it with isinstance, not .is_ok() — in result 0.17 the predicates are annotated Literal[True]/Literal[False] rather than TypeGuard, so they don't narrow for a type checker.

from result import Ok, Err
from pocketbase.utils.errors import ErrorType

res = await pb.collection("posts").get_one(post_id)
if isinstance(res, Err):
    exc = res.err_value
    if exc.type == ErrorType.RECORD_NOT_FOUND:
        return None
    log.error(exc.details)     # type, message, status, method, url, input, response body
    raise exc
post = res.ok_value

Prefer exceptions in a script or a request handler? Use the bundled bridges rather than Result.unwrap(), which buries the failure in an opaque UnwrapError:

from pocketbase import unwrap, aunwrap

post = unwrap(await pb.collection("posts").get_one(post_id))
post = await aunwrap(pb.collection("posts").get_one(post_id))   # same, one call

Both re-raise the original PocketbaseException, so .type, .details and except PocketbaseException keep working.

Records

from pocketbase.utils.params import ParamsList, ParamsOne

await pb.collection("posts").get_one("REC_ID", ParamsOne(expand="author"))
await pb.collection("posts").create({"title": "Hello", "author": user_id})
await pb.collection("posts").update("REC_ID", {"title": "Edited"})
await pb.collection("posts").delete("REC_ID")

list() returns one page. For the whole set, let the client walk the pages:

res = await pb.collection("posts").get_full_list(
    ParamsList(filter='published = true', sort="-created"), batch=500
)

get_full_list owns the pagination — page/per_page on the params are ignored — and a failed page returns its Err rather than a partial list, so a short read can't pass for a complete one.

Filters

Filter expressions are strings, so interpolating a value breaks on the first apostrophe, and a value from a request can rewrite the filter's meaning. Bind values through placeholders:

res = await pb.collection("posts").list(ParamsList(
    filter=pb.filter(
        "title ~ {:q} && author = {:author} && created >= {:since}",
        q=search_box, author=user_id, since=cutoff,
    )
))

Strings come back quoted with quotes and backslashes escaped; datetime/date render in the Y-m-d H:i:s.uZ form PocketBase compares against; numbers and booleans stay bare; None becomes null; lists and dicts become quoted JSON. A placeholder with no argument raises instead of reaching the server as syntax.

The same expression language drives collection API rules — learn it once and it applies to both.

Auth

auth = await pb.collection("users").auth_with_password("jane@example.com", "secret")
token = auth.ok_value.token          # persist to reuse the session elsewhere
user  = auth.ok_value.record

pb2 = Pocketbase(url=..., token=token)   # or pb2.set_token(token)

await pb.superusers.auth_with_password("admin@example.com", "secret")
await pb.superusers.impersonate(user_id, duration=600)

Also on any auth collection: auth_refresh, request_verification, confirm_verification, request_password_reset, confirm_password_reset, request_email_change, confirm_email_change.

Collections

from pocketbase.models.collection.create import (
    CreateCollection, CreateTextField, CreateSelectField,
    CreateRelationField, CreateAutoDateField,
)

await pb.superusers.auth_with_password("admin@example.com", "secret")

res = await pb.collections.create(CreateCollection(
    name="posts",
    type="base",
    fields=[
        CreateTextField(name="title", required=True, min=3, max=200),
        CreateSelectField(name="status", values=["draft", "published"], max_select=1),
        CreateRelationField(name="author", collectionId="_pb_users_auth_", maxSelect=1),
        CreateAutoDateField(name="created", on_create=True, on_update=False),
        CreateAutoDateField(name="updated", on_create=True, on_update=True),
    ],
    list_rule='@request.auth.id != ""',
    view_rule="",
    create_rule='@request.auth.id != ""',
    update_rule="author = @request.auth.id",
    delete_rule="author = @request.auth.id",
    indexes=["CREATE UNIQUE INDEX idx_posts_title ON posts (title)"],
))

Every field option accepts both its Python name and its camelCase wire name — max_select or maxSelect, on_update or onUpdate — and an unknown keyword raises rather than being silently ignored. A rule of "" means anyone including guests; None means superusers only.

Also available: get_one, update, delete, truncate, import_collections, scaffolds, dry_run_view, get_full_list.

Batch

All requests run in one transaction; any failure rolls back the lot. Enable batch in PocketBase settings first.

batch = pb.create_batch()
batch.collection("posts").create({"title": "a"})     # queued, not awaited
batch.collection("posts").update("REC_ID", {"title": "b"})
batch.collection("posts").delete("OTHER_ID")
res = await batch.send()

Files

from pocketbase.utils.file_upload import FileUpload

await pb.collection("posts").create({
    "title": "With attachment",
    "document": FileUpload("/path/to/file.pdf"),
})

url = pb.files.get_url(record, "photo.png", thumb="100x100")   # no network call
res = await pb.files.download(record, "photo.png")             # Result[bytes, ...]

The request switches to multipart/form-data automatically when a FileUpload is in the body, batches included. download buffers the whole file in memory; hand get_url() to a streaming client for large ones.

Operations

Superuser-only, all returning Result:

await pb.backups.create("nightly.zip")
await pb.backups.restore(key)          # replaces the live database
await pb.logs.list(ParamsList(per_page=100))
await pb.crons.run(job_id)
await pb.settings.update({"meta": {"appName": "My App"}})
await pb.health.check()                # no auth required

Agent skill

skills/pypocketbase/ is an agent skill covering this library's idioms — the Result contract, the filter language, collection field types, and the traps worth knowing. Point a coding agent at it, or copy it into .claude/skills/.

It ships a static checker for schema code:

python skills/pypocketbase/scripts/check_collection.py bootstrap.py

License

MIT