Basics - TrigonDev/apgorm Wiki

Declaring Models

Each model should be a subclass of apgorm.Model, with columns declared on it as class vars.

import apgorm
from apgorm.types import VarChar, Serial

class User(apgorm.Model):
    name = VarChar(32).field()

    primary_key = (name,)

Note that all models must have a primary key, but the fields in the primary key can be whatever you want.

After you've declared your models, you need to create a database (which is also a subclass of apgorm.Database):

class Database(apgorm.Database):
    users = User  # sets the tablename to "users"

Setup & Migrations

The first step after declaring your models and database is to connect to your Postgres database:

db = Database("path/to/migrations/folder")
await db.connect(database="my_database_name", user="...", password="...")

It's always a good idea to close the database when you're done. This can be done with await db.cleanup().

The next step is to create and apply migrations:

if db.must_create_migrations():
if await db.must_apply_migrations():
    await db.apply_migrations()

Now, every time you run this script, the database will check if anything has changed since the last migration was created. If it has, then it will create a new migration. It also checks the currently connected database to see if the migrations need to be applied and then applies them if necessary. You can see some examples of migrations under the examples folder under <example name>/migrations.

Using the Database

Now you can actually use your models.

Create a user:

user = await User(name="Me").create()
print(user)  # -> <User name:"Me">

Change the user name: = "New Me"
print(user)  # -> <User name:"New Me">

Fetch the user:

user_copy = await User.fetch(name="New Me")
print(user_copy)  # -> <User name:"New Me">

Note that "is" test fails between model copies:

print(user_copy is user)  # -> False

But == tests work:

print(user_copy == user)  # -> True

Also, updating one model won't update copies of it: = "New New Me"
print(user_copy, user) # -> <User name:"New New Me"> <User name:"New Me">

Until you refetch:

await user.refetch()
print(user)  # -> <User name:"New New Me">

Deleting a row is done with .delete():

await user.delete()