ormar 0.3.6

An simple async ORM with fastapi in mind and pydantic validation.

ORMar

The ormar package is an async ORM for Python, with support for Postgres, MySQL, and SQLite.

Ormar - apart form obvious ORM in name - get it's name from ormar in swedish which means snakes, and ormar(e) in italian which means cabinet. And what's a better name for python ORM than snakes cabinet :)

Ormar is built with:

• SQLAlchemy core for query building.
• databases for cross-database async support.
• pydantic for data validation.

Because ormar is built on SQLAlchemy core, you can use alembic to provide database migrations.

The goal was to create a simple ORM that can be used directly (as request and response models) with fastapi that bases it's data validation on pydantic. Initial work was inspired by encode/orm, later I found ormantic and used it as a further inspiration. The encode package was too simple (i.e. no ability to join two times to the same table) and used typesystem for data checks.

ormar is still under development: We recommend pinning any dependencies with ormar~=0.2.0

Note: Use ipython to try this from the console, since it supports await.

import databases
import ormar
import sqlalchemy

database = databases.Database("sqlite:///db.sqlite")
metadata = sqlalchemy.MetaData()

class Note(ormar.Model):
    class Meta:
        tablename = "notes"
        database = database
        metadata = metadata

    # primary keys of type int by dafault are set to autoincrement    
    id: ormar.Integer(primary_key=True)
    text: ormar.String(length=100)
    completed: ormar.Boolean(default=False)
    # as of ormar >=0.3.2 you can provide a list of choices that will be validated
    flag: ormar.String(default='To do', choices=['To do', 'Pending', 'Done'])

# Create the database
engine = sqlalchemy.create_engine(str(database.url))
metadata.create_all(engine)

# .create()
await Note.objects.create(text="Buy the groceries.", completed=False)
await Note.objects.create(text="Call Mum.", completed=True)
await Note.objects.create(text="Send invoices.", completed=True)

# .all()
notes = await Note.objects.all()

# .filter()
notes = await Note.objects.filter(completed=True).all()

# exact, iexact, contains, icontains, lt, lte, gt, gte, in
notes = await Note.objects.filter(text__icontains="mum").all()

# exclude - from ormar >= 0.3.1
notes = await Note.objects.exclude(text__icontains="mum").all()

# startswith, istartswith, endswith, iendswith - from ormar >= 0.3.3
notes = await Note.objects.filter(text__iendswith="mum.").all()
notes = await Note.objects.filter(text__istartswith="call").all()
notes = await Note.objects.filter(text__startswith="Buy").all()

# .get()
note = await Note.objects.get(id=1)

# .update()
await note.update(completed=True)

# .delete()
await note.delete()

# 'pk' always refers to the primary key
note = await Note.objects.get(pk=2)
note.pk  # 2

Ormar supports loading and filtering across foreign keys...

import databases
import ormar
import sqlalchemy

database = databases.Database("sqlite:///db.sqlite")
metadata = sqlalchemy.MetaData()


class Album(ormar.Model):
    class Meta:
        tablename = "album"
        metadata = metadata
        database = database

    id: ormar.Integer(primary_key=True)
    name: ormar.String(length=100)


class Track(ormar.Model):
    class Meta:
        tablename = "track"
        metadata = metadata
        database = database

    id: ormar.Integer(primary_key=True)
    album: ormar.ForeignKey(Album)
    title: ormar.String(length=100)
    position: ormar.Integer()


# Create some records to work with.
malibu = await Album.objects.create(name="Malibu")
await Track.objects.create(album=malibu, title="The Bird", position=1)
await Track.objects.create(album=malibu, title="Heart don't stand a chance", position=2)
await Track.objects.create(album=malibu, title="The Waters", position=3)

fantasies = await Album.objects.create(name="Fantasies")
await Track.objects.create(album=fantasies, title="Help I'm Alive", position=1)
await Track.objects.create(album=fantasies, title="Sick Muse", position=2)


# Fetch an instance, without loading a foreign key relationship on it.
track = await Track.objects.get(title="The Bird")

# We have an album instance, but it only has the primary key populated
print(track.album)       # Album(id=1) [sparse]
print(track.album.pk)    # 1
print(track.album.name)  # Raises AttributeError

# Load the relationship from the database
await track.album.load()
assert track.album.name == "Malibu"

# This time, fetch an instance, loading the foreign key relationship.
track = await Track.objects.select_related("album").get(title="The Bird")
assert track.album.name == "Malibu"

# By default you also get a second side of the relation 
# constructed as lowercase source model name +'s' (tracks in this case)
# you can also provide custom name with parameter related_name
album = await Album.objects.select_related("tracks").all()
assert len(album.tracks) == 3

# Fetch instances, with a filter across an FK relationship.
tracks = Track.objects.filter(album__name="Fantasies")
assert len(tracks) == 2

# Fetch instances, with a filter and operator across an FK relationship.
tracks = Track.objects.filter(album__name__iexact="fantasies")
assert len(tracks) == 2

# Limit a query
tracks = await Track.objects.limit(1).all()
assert len(tracks) == 1

Since version >=0.3 Ormar supports also many to many relationships

import databases
import ormar
import sqlalchemy

database = databases.Database("sqlite:///db.sqlite")
metadata = sqlalchemy.MetaData()

class Author(ormar.Model):
    class Meta:
        tablename = "authors"
        database = database
        metadata = metadata

    id: ormar.Integer(primary_key=True)
    first_name: ormar.String(max_length=80)
    last_name: ormar.String(max_length=80)


class Category(ormar.Model):
    class Meta:
        tablename = "categories"
        database = database
        metadata = metadata

    id: ormar.Integer(primary_key=True)
    name: ormar.String(max_length=40)


class PostCategory(ormar.Model):
    class Meta:
        tablename = "posts_categories"
        database = database
        metadata = metadata


class Post(ormar.Model):
    class Meta:
        tablename = "posts"
        database = database
        metadata = metadata

    id: ormar.Integer(primary_key=True)
    title: ormar.String(max_length=200)
    categories: ormar.ManyToMany(Category, through=PostCategory)
    author: ormar.ForeignKey(Author)

guido = await Author.objects.create(first_name="Guido", last_name="Van Rossum")
post = await Post.objects.create(title="Hello, M2M", author=guido)
news = await Category.objects.create(name="News")

# Add a category to a post.
await post.categories.add(news)
# or from the other end:
await news.posts.add(post)

# Creating columns object from instance:
await post.categories.create(name="Tips")
assert len(await post.categories.all()) == 2

# Many to many relation exposes a list of columns models 
# and an API of the Queryset:
assert news == await post.categories.get(name="News")

# with all Queryset methods - filtering, selecting columns, counting etc.
await news.posts.filter(title__contains="M2M").all()
await Category.objects.filter(posts__author=guido).get()

# columns models of many to many relation can be prefetched
news_posts = await news.posts.select_related("author").all()
assert news_posts[0].author == guido

# Removal of the relationship by one
await news.posts.remove(post)
# or all at once
await news.posts.clear()

Since version >=0.3.4 Ormar supports also queryset level delete and update statements, as well as get_or_create and update_or_create

import databases
import ormar
import sqlalchemy

database = databases.Database("sqlite:///db.sqlite")
metadata = sqlalchemy.MetaData()

class Book(ormar.Model):
    class Meta:
        tablename = "books"
        metadata = metadata
        database = database

    id: ormar.Integer(primary_key=True)
    title: ormar.String(max_length=200)
    author: ormar.String(max_length=100)
    genre: ormar.String(max_length=100, default='Fiction', choices=['Fiction', 'Adventure', 'Historic', 'Fantasy'])

await Book.objects.create(title='Tom Sawyer', author="Twain, Mark", genre='Adventure')
await Book.objects.create(title='War and Peace', author="Tolstoy, Leo", genre='Fiction')
await Book.objects.create(title='Anna Karenina', author="Tolstoy, Leo", genre='Fiction')
await Book.objects.create(title='Harry Potter', author="Rowling, J.K.", genre='Fantasy')
await Book.objects.create(title='Lord of the Rings', author="Tolkien, J.R.", genre='Fantasy')

# update accepts kwargs that are used to update queryset model
# all other arguments are ignored (argument names not in own model table)
await Book.objects.filter(author="Tolstoy, Leo").update(author="Lenin, Vladimir") # update all Tolstoy's books
all_books = await Book.objects.filter(author="Lenin, Vladimir").all()
assert len(all_books) == 2

# delete accepts kwargs that will be used in filter
# acting in same way as queryset.filter(**kwargs).delete()
await Book.objects.delete(genre='Fantasy') # delete all fantasy books
all_books = await Book.objects.all()
assert len(all_books) == 3

# queryset needs to be filtered before deleting to prevent accidental overwrite
# to update whole database table each=True needs to be provided as a safety switch
await Book.objects.update(each=True, genre='Fiction')
all_books = await Book.objects.filter(genre='Fiction').all()
assert len(all_books) == 3

# helper get/update or create methods of queryset
# if not exists it will be created
vol1 = await Book.objects.get_or_create(title="Volume I", author='Anonymous', genre='Fiction')
assert await Book.objects.count() == 1

# if exists it will be returned
assert await Book.objects.get_or_create(title="Volume I", author='Anonymous', genre='Fiction') == vol1
assert await Book.objects.count() == 1

# if not exist the instance will be persisted in db
vol2 = await Book.objects.update_or_create(title="Volume II", author='Anonymous', genre='Fiction')
assert await Book.objects.count() == 1

# if pk or pkname passed in kwargs (like id here) the object will be updated
assert await Book.objects.update_or_create(id=vol2.id, genre='Historic')
assert await Book.objects.count() == 1

Since version >=0.3.5 Ormar supports also bulk operations -> bulk_create and bulk_update

import databases
import ormar
import sqlalchemy

database = databases.Database("sqlite:///db.sqlite")
metadata = sqlalchemy.MetaData()


class ToDo(ormar.Model):
    class Meta:
        tablename = "todos"
        metadata = metadata
        database = database

    id: ormar.Integer(primary_key=True)
    text: ormar.String(max_length=500)
    completed: ormar.Boolean(default=False)

# create multiple instances at once with bulk_create
await ToDo.objects.bulk_create(
            [
                ToDo(text="Buy the groceries."),
                ToDo(text="Call Mum.", completed=True),
                ToDo(text="Send invoices.", completed=True),
            ]
        )

todoes = await ToDo.objects.all()
assert len(todoes) == 3

# update objects
for todo in todoes:
    todo.completed = False

# perform update of all objects at once
# objects need to have pk column set, otherwise exception is raised
await ToDo.objects.bulk_update(todoes)

completed = await ToDo.objects.filter(completed=False).all()
assert len(completed) == 3

Since version >=0.3.6 Ormar supports unique constraints on multiple columns

import databases
import ormar
import sqlalchemy

database = databases.Database("sqlite:///db.sqlite")
metadata = sqlalchemy.MetaData()


class Product(ormar.Model):
    class Meta:
        tablename = "products"
        metadata = metadata
        database = database
        # define your constraints in Meta class of the model
        # it's a list that can contain multiple constraints
        constraints = [ormar.UniqueColumns("name", "company")]

    id: ormar.Integer(primary_key=True)
    name: ormar.String(max_length=100)
    company: ormar.String(max_length=200)

await Product.objects.create(name="Cookies", company="Nestle")
await Product.objects.create(name="Mars", company="Mars")
await Product.objects.create(name="Mars", company="Nestle")


# will raise error based on backend 
# (sqlite3.IntegrityError, pymysql.IntegrityError, asyncpg.exceptions.UniqueViolationError)
await Product.objects.create(name="Mars", company="Mars")       

Since version >=0.3.6 Ormar supports selecting subset of model columns to limit the data load. Warning - mandatory fields cannot be excluded as it will raise validation error, to exclude a field it has to be nullable. Pk column cannot be excluded - it's always auto added even if not explicitly included.

import databases
import pydantic
import pytest
import sqlalchemy

import ormar
from tests.settings import DATABASE_URL

database = databases.Database(DATABASE_URL, force_rollback=True)
metadata = sqlalchemy.MetaData()


class Company(ormar.Model):
    class Meta:
        tablename = "companies"
        metadata = metadata
        database = database

    id: ormar.Integer(primary_key=True)
    name: ormar.String(max_length=100)
    founded: ormar.Integer(nullable=True)


class Car(ormar.Model):
    class Meta:
        tablename = "cars"
        metadata = metadata
        database = database

    id: ormar.Integer(primary_key=True)
    manufacturer: ormar.ForeignKey(Company)
    name: ormar.String(max_length=100)
    year: ormar.Integer(nullable=True)
    gearbox_type: ormar.String(max_length=20, nullable=True)
    gears: ormar.Integer(nullable=True)
    aircon_type: ormar.String(max_length=20, nullable=True)



# build some sample data
toyota = await Company.objects.create(name="Toyota", founded=1937)
await Car.objects.create(manufacturer=toyota, name="Corolla", year=2020, gearbox_type='Manual', gears=5,
                         aircon_type='Manual')
await Car.objects.create(manufacturer=toyota, name="Yaris", year=2019, gearbox_type='Manual', gears=5,
                         aircon_type='Manual')
await Car.objects.create(manufacturer=toyota, name="Supreme", year=2020, gearbox_type='Auto', gears=6,
                         aircon_type='Auto')

# select manufacturer but only name - to include related models use notation {model_name}__{column}
all_cars = await Car.objects.select_related('manufacturer').fields(['id', 'name', 'company__name']).all()
for car in all_cars:
    # excluded columns will yield None
    assert all(getattr(car, x) is None for x in ['year', 'gearbox_type', 'gears', 'aircon_type'])
    # included column on related models will be available, pk column is always included
    # even if you do not include it in fields list
    assert car.manufacturer.name == 'Toyota'
    # also in the nested related models - you cannot exclude pk - it's always auto added 
    assert car.manufacturer.founded is None

# fields() can be called several times, building up the columns to select
# models selected in select_related but with no columns in fields list implies all fields
all_cars = await Car.objects.select_related('manufacturer').fields('id').fields(
                ['name']).all()
# all fiels from company model are selected
assert all_cars[0].manufacturer.name == 'Toyota' 
assert all_cars[0].manufacturer.founded ==  1937

# cannot exclude mandatory model columns - company__name in this example
await Car.objects.select_related('manufacturer').fields(['id', 'name', 'company__founded']).all()
# will raise pydantic ValidationError as company.name is required

Data types

The following keyword arguments are supported on all field types.

• primary_key: bool
• nullable: bool
• default: Any
• server_default: Any
• index: bool
• unique: bool
• choices: typing.Sequence

All fields are required unless one of the following is set:

• nullable - Creates a nullable column. Sets the default to None.
• default - Set a default value for the field.
• server_default - Set a default value for the field on server side (like sqlalchemy's func.now()).
• primary key with autoincrement - When a column is set to primary key and autoincrement is set on this column. Autoincrement is set by default on int primary keys.

Available Model Fields (with required args - optional ones in docs):

• String(max_length)
• Text()
• Boolean()
• Integer()
• Float()
• Date()
• Time()
• DateTime()
• JSON()
• BigInteger()
• Decimal(scale, precision)
• UUID()
• ForeignKey(to)
• Many2Many(to, through)