Skip to content

p3orm

p3orm logo

Utilitarian Python ORM for Postgres/SQLite powered by asyncpg/aiosqlite, Pydantic, and PyPika


Documentation: https://rafalstapinski.github.io/p3orm

Source Code: https://github.com/rafalstapinski/p3orm


Test Status pypi Supported Python Versions: 3.8, 3.9, 3.10 MIT License


Philosophy

90% of the time we talk to a database is with a CRUD operation. p3orm provides convenience helpers for fetching (one, first, many), inserting (one, many), updating (one), and deleting (one, many).

The remaining 10% is a bit more complicated. p3orm doesn't attempt to hide SQL queries or database interactions behind any magic. Instead, it empowers you to write direct and legible SQL queries with PyPika and execute them explicitly against the database.

Notably, objects created or fetched by p3orm are dead, they're just Pydantic models. If you want to interact with the database, you do so explicitly.

tl;dr - p3orm makes easy things easy, and hard things possible


Features

  • Comprehensive type annotations (full intellisense support)
  • String type validation an parsing powered by Pydantic
  • Support for PyPika queries
  • Support for all postgres datatypes
  • Support for all sqlite datatypes

Installation

Install with poetry

poetry add p3orm[sqlite]
# or
poetry add p3orm[postgres]

or with pip

pip install p3orm[sqlite]
# or
pip install p3orm[postgres]

The [sqlite] extra installs aiosqlite as p3orm's database driver, whereas [postgres] installs asyncpg.


Basic Usage

from datetime import datetime
from p3orm import Column, Table

from p3orm import sqlite as db
# or: from p3orm import postgres as db

class Thing(Table):
    id = Column(int, pk=True, autogen=True)
    name = Column(str)
    created_at = Column(datetime, autogen=True)

await db().connect(":memory:")

thing = Thing(name="Name")

inserted = await Thing.insert_one(thing)

fetched = await Thing.fetch_first(Thing.id == 1)

fetched.name = "Changed"

updated = await Thing.update_one(fetched)

deleted = await Thing.delete_where(Thing.id == updated.id)

await db().disconnect()