p3orm 0.10.3

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

p3orm

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

---

---

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()