Skip to content

Getting started

Connecting to a database

Postgres

from p3orm import postgres

db = postgres()

await db.connect("postgresql://user:pass@host:port")

# or

await db.connect(user=..., password=..., database=..., host=..., port=...)

db.is_connected() # False

SQLite

from p3orm import sqlite

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

Disconnecting

Postgres

await postgres().disconnect()

postgres().is_connected() # False

SQLite

await sqlite().disconnect()

p3orm currently allows you to connect to a single database within a process. You can fetch the Postgres driver with postgres() from p3orm.core and the SQLite driver with sqlite() from p3orm.core.

Defining models

from datetime import datetime

from p3orm import Table, Column

class Thing(Table):
  __tablename__ = "thing"

  id = Column(int, autogen=True)
  name = Column(Optional[str])
  created_at = Column(datetime, autogen=True)

Column parameters

  • _type: Specifies the type of the column, will be used to convert between Python and Postgres. This also generates the type annotation for an instance of your model. For a list of supported types, see the asyncpg docs.
  • name: Specifies the name of the column. This (currently) has to match the name of the field on the class as well.
  • pk: Specifies whether this is the primary key. This is used for the update_one convenience function.
  • autogen: Specifies whether Postgres will automatically generate/update the value of this field, e.g. SERIAL for ids or DEFAULT CURRENT_TIMESTAMP for created_at or updated_at type fields.

Basic queries

p3orm.table.Table provides the following convenience methods for executing queries with basic criteria.

  • Table.fetch_one will raise a MultipleResultsReturned or NoResultsReturned if exactly one result isn't returned.
  • Table.fetch_first will return the first result from a query or None.
  • Table.fetch_all will return a list of results (or an empty list) of a query.
  • Table.insert_one will insert exactly one model.
  • Table.insert_many will insert multiple models with a performant bulk insert query.
  • Table.update_one will accept a single model and update it based on its primary key.
  • Table.delete_where will delete and return all rows that match a criterion.

All of the above methods use parameterized queries under the hood to protect against SQL injections.

Creating

inserted_result = await Thing.insert_one(Thing(name="thing 1"))

to_insert = [Thing(name="thing 2"), Thing(name="thing 3")]

inserted_results = await Thing.insert_many(to_insert)

Note that we didn't specify the id or created_at fields. Those are autogenerated by Postgres, so we let the database do the work. We also didn't explicitly specify whether the fields are nullable or not. Again, we offload all validation to the database as the source of truth, and only specify the Optional type for local type checking and intellisense. It is best practice to discard locally created models after they've been inserted and replace them with the returned set as they will contain all the database-generated and validated field values.

Retrieving

everything = await Thing.fetch_all()

filtered = await Thing.fetch_all(Thing.id > 2)

first = await Thing.fetch_first(Thing.id > 2)

one = await Thing.fetch_one(Thing.id == 1)

Updating

thing = await Thing.fetch_one(Thing.id == 2)

thing.name = "Changed"

updated_thing = await Thing.update_one(thing)

It is best practice to discard the local model after an update and replace it with the updated result as it will contain any database-updated fields.

Deleting

deleted_things = await Thing.delete_where(Thing.id > 10)
Table.delete_where always returns a list of the deleted records.