Skip to content

Queries

QuerySet

Each Model is auto registered with a QuerySet that represents the underlaying query and it's options.

Most of the methods are also available through many to many relation interface.

Given the Models like this

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
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(max_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(max_length=100)
    position: ormar.Integer()

we can demonstrate available methods to fetch and save the data into the database.

create

create(**kwargs): -> Model

Creates the model instance, saves it in a database and returns the updates model (with pk populated if not passed and autoincrement is set).

The allowed kwargs are Model fields names and proper value types. 

1
2
malibu = await Album.objects.create(name="Malibu")
await Track.objects.create(album=malibu, title="The Bird", position=1)

The alternative is a split creation and persistence of the Model. 

1
2
malibu = Album(name="Malibu")
await malibu.save()

Tip

Check other Model methods in models

get

get(**kwargs): -> Model

Get's the first row from the db meeting the criteria set by kwargs.

If no criteria set it will return the first row in db.

Passing a criteria is actually calling filter(**kwargs) method described below.

1
2
3
4
track = await Track.objects.get(name='The Bird')
# note that above is equivalent to await Track.objects.filter(name='The Bird').get()
track2 = track = await Track.objects.get()
track == track2 # True since it's the only row in db in our example

Warning

If no row meets the criteria NoMatch exception is raised.

If there are multiple rows meeting the criteria the MultipleMatches exception is raised.

get_or_create

get_or_create(**kwargs) -> Model

Combination of create and get methods.

Tries to get a row meeting the criteria and if NoMatch exception is raised it creates a new one with given kwargs.

1
2
3
4
5
album = await Album.objects.get_or_create(name='The Cat')
# object is created as it does not exist
album2 = await Album.objects.get_or_create(name='The Cat')
assert album == album2
# return True as the same db row is returned

Warning

Despite being a equivalent row from database the album and album2 in example above are 2 different python objects! Updating one of them will not refresh the second one until you excplicitly load() the fresh data from db.

Note

Note that if you want to create a new object you either have to pass pk column value or pk column has to be set as autoincrement

update

update(each: bool = False, **kwargs) -> int

QuerySet level update is used to update multiple records with the same value at once.

You either have to filter the QuerySet first or provide a each=True flag to update whole table.

If you do not provide this flag or a filter a QueryDefinitionError will be raised.

Return number of rows updated.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
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')


# 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

update_or_create

update_or_create(**kwargs) -> Model

Updates the model, or in case there is no match in database creates a new one.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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')


# 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

Note

Note that if you want to create a new object you either have to pass pk column value or pk column has to be set as autoincrement

bulk_create

bulk_create(objects: List["Model"]) -> None

Allows you to create multiple objects at once.

A valid list of Model objects needs to be passed.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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

bulk_update

bulk_update(objects: List["Model"], columns: List[str] = None) -> None

Allows to update multiple instance at once.

All Models passed need to have primary key column populated.

You can also select which fields to update by passing columns list as a list of string names.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
# continuing the example from bulk_create
# 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

delete

delete(each: bool = False, **kwargs) -> int

QuerySet level delete is used to delete multiple records at once.

You either have to filter the QuerySet first or provide a each=True flag to delete whole table.

If you do not provide this flag or a filter a QueryDefinitionError will be raised.

Return number of rows deleted.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
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 in Space', author="Tolstoy, Leo", genre='Fantasy')
await Book.objects.create(title='Anna Karenina', author="Tolstoy, Leo", genre='Fiction')

# 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) == 2

all

all(self, **kwargs) -> List[Optional["Model"]]

Returns all rows from a database for given model for set filter options.

Passing kwargs is a shortcut and equals to calling filter(**kwrags).all().

If there are no rows meeting the criteria an empty list is returned.

1
2
3
4
5
tracks = await Track.objects.select_related("album").all(title='Sample')
# will return a list of all Tracks with title Sample

tracks = await Track.objects.all()
# will return a list of all Tracks in database

filter

filter(**kwargs) -> QuerySet

Allows you to filter by any Model attribute/field as well as to fetch instances, with a filter across an FK relationship.

1
2
3
4
5
track = Track.objects.filter(name="The Bird").get()
# will return a track with name equal to 'The Bird'

tracks = Track.objects.filter(album__name="Fantasies").all()
# will return all tracks where the columns album name = 'Fantasies'

You can use special filter suffix to change the filter operands:

  • exact - like album__name__exact='Malibu' (exact match)
  • iexact - like album__name__iexact='malibu' (exact match case insensitive)
  • contains - like album__name__conatins='Mal' (sql like)
  • icontains - like album__name__icontains='mal' (sql like case insensitive)
  • in - like album__name__in=['Malibu', 'Barclay'] (sql in)
  • gt - like position__gt=3 (sql >)
  • gte - like position__gte=3 (sql >=)
  • lt - like position__lt=3 (sql <)
  • lte - like position__lte=3 (sql <=)
  • startswith - like album__name__startswith='Mal' (exact start match)
  • istartswith - like album__name__istartswith='mal' (exact start match case insensitive)
  • endswith - like album__name__endswith='ibu' (exact end match)
  • iendswith - like album__name__iendswith='IBU' (exact end match case insensitive)

Note

All methods that do not return the rows explicitly returns a QueySet instance so you can chain them together

So operations like filter(), select_related(), limit() and offset() etc. can be chained.

Something like Track.object.select_related("album").filter(album__name="Malibu").offset(1).limit(1).all()

exclude

exclude(**kwargs) -> QuerySet

Works exactly the same as filter and all modifiers (suffixes) are the same, but returns a not condition.

So if you use filter(name='John') which equals to where name = 'John' in SQL, the exclude(name='John') equals to where name <> 'John'

Note that all conditions are joined so if you pass multiple values it becomes a union of conditions.

exclude(name='John', age>=35) will become where not (name='John' and age>=35)

1
2
notes = await Track.objects.exclude(position_gt=3).all()
# returns all tracks with position < 3

select_related(related: Union[List, str]) -> QuerySet

Allows to prefetch related models.

To fetch related model use ForeignKey names.

To chain related Models relation use double underscore.

1
2
album = await Album.objects.select_related("tracks").all()
# will return album will all columns tracks

You can provide a string or a list of strings

1
2
3
4
classes = await SchoolClass.objects.select_related(
["teachers__category", "students"]).all()
# will return classes with teachers and teachers categories
# as well as classes students

Exactly the same behavior is for Many2Many fields, where you put the names of Many2Many fields and the final Models are fetched for you.

Warning

If you set ForeignKey field as not nullable (so required) during all queries the not nullable Models will be auto prefetched, even if you do not include them in select_related.

Note

All methods that do not return the rows explicitly returns a QueySet instance so you can chain them together

So operations like filter(), select_related(), limit() and offset() etc. can be chained.

Something like Track.object.select_related("album").filter(album__name="Malibu").offset(1).limit(1).all()

limit

limit(limit_count: int) -> QuerySet

You can limit the results to desired number of rows.

1
2
tracks = await Track.objects.limit(1).all()
# will return just one Track

Note

All methods that do not return the rows explicitly returns a QueySet instance so you can chain them together

So operations like filter(), select_related(), limit() and offset() etc. can be chained.

Something like Track.object.select_related("album").filter(album__name="Malibu").offset(1).limit(1).all()

offset

offset(offset: int) -> QuerySet

You can also offset the results by desired number of rows.

1
2
tracks = await Track.objects.offset(1).limit(1).all()
# will return just one Track, but this time the second one

Note

All methods that do not return the rows explicitly returns a QueySet instance so you can chain them together

So operations like filter(), select_related(), limit() and offset() etc. can be chained.

Something like Track.object.select_related("album").filter(album__name="Malibu").offset(1).limit(1).all()

count

count() -> int

Returns number of rows matching the given criteria (applied with filter and exclude)

1
2
# returns count of rows in db
no_of_books = await Book.objects.count()

exists

exists() -> bool

Returns a bool value to confirm if there are rows matching the given criteria (applied with filter and exclude)

1
2
# returns a boolean value if given row exists
has_sample = await Book.objects.filter(title='Sample').exists()

fields

fields(columns: Union[List, str]) -> QuerySet

With fields() you can select subset of model columns to limit the data load.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
import databases
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

Warning

Mandatory fields cannot be excluded as it will raise ValidationError, to exclude a field it has to be nullable.

Tip

Pk column cannot be excluded - it's always auto added even if not explicitly included.

Note

All methods that do not return the rows explicitly returns a QueySet instance so you can chain them together

So operations like filter(), select_related(), limit() and offset() etc. can be chained.

Something like Track.object.select_related("album").filter(album__name="Malibu").offset(1).limit(1).all()