# note this is just a partial snippet full working example below# 1. Importsimportsqlalchemyimportdatabases# 2. InitializationDATABASE_URL="sqlite:///db.sqlite"database=databases.Database(DATABASE_URL) metadata=sqlalchemy.MetaData() # Define models here# 3. Database creation and tables creationengine=sqlalchemy.create_engine(DATABASE_URL) metadata.create_all(engine)

fromtypingimportOptionalimportdatabasesimportpydanticimportormarimportsqlalchemyDATABASE_URL="sqlite:///db.sqlite"base_ormar_config=ormar.OrmarConfig( database=databases.Database(DATABASE_URL), metadata=sqlalchemy.MetaData(), engine=sqlalchemy.create_engine(DATABASE_URL), ) # Note that all type hints are optional# below is a perfectly valid model declaration# class Author(ormar.Model):# ormar_config = base_ormar_config.copy(tablename="authors")## id = ormar.Integer(primary_key=True) # <= notice no field types# name = ormar.String(max_length=100)classAuthor(ormar.Model): ormar_config=base_ormar_config.copy(tablename="authors") id: int=ormar.Integer(primary_key=True) name: str=ormar.String(max_length=100) classBook(ormar.Model): ormar_config=base_ormar_config.copy(tablename="books") id: int=ormar.Integer(primary_key=True) author: Optional[Author] =ormar.ForeignKey(Author) title: str=ormar.String(max_length=100) year: int=ormar.Integer(nullable=True) # create the database# note that in production you should use migrations# note that this is not required if you connect to existing database# just to be sure we clear the db beforebase_ormar_config.metadata.drop_all(base_ormar_config.engine) base_ormar_config.metadata.create_all(base_ormar_config.engine) # all functions below are divided into functionality categories# note how all functions are defined with async - hence can use await AND needs to# be awaited on their ownasyncdefcreate(): # Create some records to work with through QuerySet.create method.# Note that queryset is exposed on each Model's class as objectstolkien=awaitAuthor.objects.create(name="J.R.R. Tolkien") awaitBook.objects.create(author=tolkien, title="The Hobbit", year=1937) awaitBook.objects.create(author=tolkien, title="The Lord of the Rings", year=1955) awaitBook.objects.create(author=tolkien, title="The Silmarillion", year=1977) # alternative creation of object divided into 2 stepssapkowski=Author(name="Andrzej Sapkowski") # do some stuffawaitsapkowski.save() # or save() after initializationawaitBook(author=sapkowski, title="The Witcher", year=1990).save() awaitBook(author=sapkowski, title="The Tower of Fools", year=2002).save() # to read more about inserting data into the database# visit: https://collerek.github.io/ormar/queries/create/asyncdefread(): # Fetch an instance, without loading a foreign key relationship on it.# Django stylebook=awaitBook.objects.get(title="The Hobbit") # or python stylebook=awaitBook.objects.get(Book.title=="The Hobbit") book2=awaitBook.objects.first() # first() fetch the instance with lower primary key valueassertbook==book2# you can access all fields on loaded modelassertbook.title=="The Hobbit"assertbook.year==1937# when no condition is passed to get()# it behaves as last() based on primary key columnbook3=awaitBook.objects.get() assertbook3.title=="The Tower of Fools"# When you have a relation, ormar always defines a related model for you# even when all you loaded is a foreign key value like in this exampleassertisinstance(book.author, Author) # primary key is populated from foreign key stored in books tableassertbook.author.pk==1# since the related model was not loaded all other fields are Noneassertbook.author.nameisNone# Load the relationship from the database when you already have the related model# alternatively see joins section belowawaitbook.author.load() assertbook.author.name=="J.R.R. Tolkien"# get all rows for given modelauthors=awaitAuthor.objects.all() assertlen(authors) ==2# to read more about reading data from the database# visit: https://collerek.github.io/ormar/queries/read/asyncdefupdate(): # read existing row from dbtolkien=awaitAuthor.objects.get(name="J.R.R. Tolkien") asserttolkien.name=="J.R.R. Tolkien"tolkien_id=tolkien.id# change the selected propertytolkien.name="John Ronald Reuel Tolkien"# call update on a model instanceawaittolkien.update() # confirm that object was updatedtolkien=awaitAuthor.objects.get(name="John Ronald Reuel Tolkien") asserttolkien.name=="John Ronald Reuel Tolkien"asserttolkien.id==tolkien_id# alternatively update data without loadingawaitAuthor.objects.filter(name__contains="Tolkien").update(name="J.R.R. Tolkien") # to read more about updating data in the database# visit: https://collerek.github.io/ormar/queries/update/asyncdefdelete(): silmarillion=awaitBook.objects.get(year=1977) # call delete() on instanceawaitsilmarillion.delete() # alternatively delete without loadingawaitBook.objects.delete(title="The Tower of Fools") # note that when there is no record ormar raises NoMatch exceptiontry: awaitBook.objects.get(year=1977) exceptormar.NoMatch: print("No book from 1977!") # to read more about deleting data from the database# visit: https://collerek.github.io/ormar/queries/delete/# note that despite the fact that record no longer exists in database# the object above is still accessible and you can use it (and i.e. save()) again.tolkien=silmarillion.authorawaitBook.objects.create(author=tolkien, title="The Silmarillion", year=1977) asyncdefjoins(): # Tho join two models use select_related# Django stylebook=awaitBook.objects.select_related("author").get(title="The Hobbit") # Python stylebook=awaitBook.objects.select_related(Book.author).get( Book.title=="The Hobbit" ) # now the author is already prefetchedassertbook.author.name=="J.R.R. Tolkien"# By default you also get a second side of the relation# constructed as lowercase source model name +'s' (books in this case)# you can also provide custom name with parameter related_name# Django styleauthor=awaitAuthor.objects.select_related("books").all(name="J.R.R. Tolkien") # Python styleauthor=awaitAuthor.objects.select_related(Author.books).all( Author.name=="J.R.R. Tolkien" ) assertlen(author[0].books) ==3# for reverse and many to many relations you can also prefetch_related# that executes a separate query for each of related models# Django styleauthor=awaitAuthor.objects.prefetch_related("books").get(name="J.R.R. Tolkien") # Python styleauthor=awaitAuthor.objects.prefetch_related(Author.books).get( Author.name=="J.R.R. Tolkien" ) assertlen(author.books) ==3# to read more about relations# visit: https://collerek.github.io/ormar/relations/# to read more about joins and subqueries# visit: https://collerek.github.io/ormar/queries/joins-and-subqueries/asyncdeffilter_and_sort(): # to filter the query you can use filter() or pass key-value pars to# get(), all() etc.# to use special methods or access related model fields use double# underscore like to filter by the name of the author use author__name# Django stylebooks=awaitBook.objects.all(author__name="J.R.R. Tolkien") # python stylebooks=awaitBook.objects.all(Book.author.name=="J.R.R. Tolkien") assertlen(books) ==3# filter can accept special methods also separated with double underscore# to issue sql query ` where authors.name like "%tolkien%"` that is not# case sensitive (hence small t in Tolkien)# Django stylebooks=awaitBook.objects.filter(author__name__icontains="tolkien").all() # python stylebooks=awaitBook.objects.filter(Book.author.name.icontains("tolkien")).all() assertlen(books) ==3# to sort use order_by() function of queryset# to sort decreasing use hyphen before the field name# same as with filter you can use double underscores to access related fields# Django stylebooks= ( awaitBook.objects.filter(author__name__icontains="tolkien") .order_by("-year") .all() ) # python stylebooks= ( awaitBook.objects.filter(Book.author.name.icontains("tolkien")) .order_by(Book.year.desc()) .all() ) assertlen(books) ==3assertbooks[0].title=="The Silmarillion"assertbooks[2].title=="The Hobbit"# to read more about filtering and ordering# visit: https://collerek.github.io/ormar/queries/filter-and-sort/asyncdefsubset_of_columns(): # to exclude some columns from loading when querying the database# you can use fields() methodhobbit=awaitBook.objects.fields(["title"]).get(title="The Hobbit") # note that fields not included in fields are empty (set to None)asserthobbit.yearisNoneasserthobbit.authorisNone# selected field is thereasserthobbit.title=="The Hobbit"# alternatively you can provide columns you want to excludehobbit=awaitBook.objects.exclude_fields(["year"]).get(title="The Hobbit") # year is still not setasserthobbit.yearisNone# but author is backasserthobbit.authorisnotNone# also you cannot exclude primary key column - it's always there# even if you EXPLICITLY exclude it it will be there# note that each model have a shortcut for primary_key column which is pk# and you can filter/access/set the values by this alias like belowasserthobbit.pkisnotNone# note that you cannot exclude fields that are not nullable# (required) in model definitiontry: awaitBook.objects.exclude_fields(["title"]).get(title="The Hobbit") exceptpydantic.ValidationError: print("Cannot exclude non nullable field title") # to read more about selecting subset of columns# visit: https://collerek.github.io/ormar/queries/select-columns/asyncdefpagination(): # to limit number of returned rows use limit()books=awaitBook.objects.limit(1).all() assertlen(books) ==1assertbooks[0].title=="The Hobbit"# to offset number of returned rows use offset()books=awaitBook.objects.limit(1).offset(1).all() assertlen(books) ==1assertbooks[0].title=="The Lord of the Rings"# alternatively use paginate that combines bothbooks=awaitBook.objects.paginate(page=2, page_size=2).all() assertlen(books) ==2# note that we removed one book of Sapkowski in delete()# and recreated The Silmarillion - by default when no order_by is set# ordering sorts by primary_key columnassertbooks[0].title=="The Witcher"assertbooks[1].title=="The Silmarillion"# to read more about pagination and number of rows# visit: https://collerek.github.io/ormar/queries/pagination-and-rows-number/asyncdefaggregations(): # count:assert2==awaitAuthor.objects.count() # existsassertawaitBook.objects.filter(title="The Hobbit").exists() # maximumassert1990==awaitBook.objects.max(columns=["year"]) # minimumassert1937==awaitBook.objects.min(columns=["year"]) # averageassert1964.75==awaitBook.objects.avg(columns=["year"]) # sumassert7859==awaitBook.objects.sum(columns=["year"]) # to read more about aggregated functions# visit: https://collerek.github.io/ormar/queries/aggregations/asyncdefraw_data(): # extract raw data in a form of dicts or tuples# note that this skips the validation(!) as models are# not created from parsed data# get list of objects as dictsassertawaitBook.objects.values() == [{"id": 1, "author": 1, "title": "The Hobbit", "year": 1937},{"id": 2, "author": 1, "title": "The Lord of the Rings", "year": 1955},{"id": 4, "author": 2, "title": "The Witcher", "year": 1990},{"id": 5, "author": 1, "title": "The Silmarillion", "year": 1977}, ] # get list of objects as tuplesassertawaitBook.objects.values_list() == [ (1, 1, "The Hobbit", 1937), (2, 1, "The Lord of the Rings", 1955), (4, 2, "The Witcher", 1990), (5, 1, "The Silmarillion", 1977), ] # filter data - note how you always get a listassertawaitBook.objects.filter(title="The Hobbit").values() == [{"id": 1, "author": 1, "title": "The Hobbit", "year": 1937} ] # select only wanted fieldsassertawaitBook.objects.filter(title="The Hobbit").values(["id", "title"]) == [{"id": 1, "title": "The Hobbit"} ] # if you select only one column you could flatten it with values_listassertawaitBook.objects.values_list("title", flatten=True) == [ "The Hobbit", "The Lord of the Rings", "The Witcher", "The Silmarillion", ] # to read more about extracting raw values# visit: https://collerek.github.io/ormar/queries/aggregations/asyncdefwith_connect(function): # note that for any other backend than sqlite you actually need to# connect to the database to perform db operationsasyncwithbase_ormar_config.database: awaitfunction() # note that if you use framework like `fastapi` you shouldn't connect# in your endpoints but have a global connection pool# check https://collerek.github.io/ormar/fastapi/ and section with db connection# gather and execute all functions# note - normally import should be at the beginning of the fileimportasyncio# note that normally you use gather() function to run several functions# concurrently but we actually modify the data and we rely on the order of functionsforfuncin [ create, read, update, delete, joins, filter_and_sort, subset_of_columns, pagination, aggregations, raw_data, ]: print(f"Executing: {func.__name__}") asyncio.run(with_connect(func)) # drop the database tablesbase_ormar_config.metadata.drop_all(base_ormar_config.engine)