The ormar package is an async mini ORM for Python, with support for Postgres,
MySQL, and SQLite.
The main benefit of using ormar are:
- getting an async ORM that can be used with async frameworks (fastapi, starlette etc.)
- getting just one model to maintain - you don't have to maintain pydantic and other orm model (sqlalchemy, peewee, gino etc.)
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.
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 :)
Check out the documentation for details.
Ormar is built with:
SQLAlchemy corefor query building.databasesfor cross-database async support.pydanticfor data validation.- typing_extensions for python 3.6 - 3.7
Because ormar is built on SQLAlchemy core, you can use alembic to provide
database migrations.
ormar is still under development: We recommend pinning any dependencies with ormar~=0.4.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 Album(ormar.Model):
class Meta:
tablename = "album"
metadata = metadata
database = database
# note that type hints are optional so
# id = ormar.Integer(primary_key=True)
# is also valid
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
class Track(ormar.Model):
class Meta:
tablename = "track"
metadata = metadata
database = database
id: int = ormar.Integer(primary_key=True)
album: Optional[Album] = ormar.ForeignKey(Album)
title: str = ormar.String(max_length=100)
position: int = 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)
# alternative creation of object divided into 2 steps
fantasies = Album.objects.create(name="Fantasies")
await fantasies.save()
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) # None
# 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) == 1create(**kwargs): -> Modelget(**kwargs): -> Modelget_or_create(**kwargs) -> Modelupdate(each: bool = False, **kwargs) -> intupdate_or_create(**kwargs) -> Modelbulk_create(objects: List[Model]) -> Nonebulk_update(objects: List[Model], columns: List[str] = None) -> Nonedelete(each: bool = False, **kwargs) -> intall(self, **kwargs) -> List[Optional[Model]]filter(**kwargs) -> QuerySetexclude(**kwargs) -> QuerySetselect_related(related: Union[List, str]) -> QuerySetlimit(limit_count: int) -> QuerySetoffset(offset: int) -> QuerySetcount() -> intexists() -> boolfields(columns: Union[List, str]) -> QuerySetexclude_fields(columns: Union[List, str]) -> QuerySetorder_by(columns:Union[List, str]) -> QuerySet
- One to many - with
ForeignKey(to: Model) - Many to many - with
ManyToMany(to: Model, through: Model)
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)ManyToMany(to, through)
The following keyword arguments are supported on all field types.
primary_key: boolnullable: booldefault: Anyserver_default: Anyindex: boolunique: boolchoices: typing.Sequencename: str
All fields are required unless one of the following is set:
nullable- Creates a nullable column. Sets the default toNone.default- Set a default value for the field.server_default- Set a default value for the field on server side (like sqlalchemy'sfunc.now()).primary keywithautoincrement- When a column is set to primary key and autoincrement is set on this column. Autoincrement is set by default on int primary keys.