Query

For demonstration convenience, let's first create several data models:

from pylite_orm import DbModel, DbField, DbType

# User model
class User(DbModel):
    id = DbField(db_type = DbType.INT, pk = True)
    name = DbField(db_type = DbType.TEXT)
    profile:'UserProfile' = RelationField(fk = 'uid')
    arts: list['Article'] = RelationField(fk = 'uid')
    created_at = DbField(db_type = DbType.TEXT, default_factory = datetime.now)

# User profile model
class UserProfile(DbModel):
    id = DbField(db_type = DbType.INT, pk = True)
    uid = DbField(db_type = DbType.INT, relation = 'user.id', on_delete = OnDelete.CASCADE)
    age = DbField(db_type = DbType.INT)
    sex = DbField(db_type = DbType.INT, default = 1)
    weight = DbField(db_type = DbType.REAL)
    user:User = RelationField(fk = 'uid')

# Article model
class Article(DbModel):
    id = DbField(db_type = DbType.INT, pk = True)
    title = DbField(db_type = DbType.TEXT)
    content = DbField(db_type = DbType.TEXT)
    uid = DbField(db_type = DbType.INT, relation = 'user.id', on_delete = OnDelete.SET_NULL)
    author:User = RelationField(fk = 'uid')

Each user has 0 or 1 profile and has published 0 or N articles. So there's a one-to-one relationship between User and UserProfile, and a one-to-many relationship between User and Article. If you use the database migration tool, three tables will be generated in the database: user, user_profile, and article.

Assume that several records have been generated for User, UserProfile, and Article through the session instance's insert method (i.e., data has been inserted into the user, user_profile, and article tables).

Queries are performed through the session instance's select method, so all the following queries are based on the dbs instance:

my_db = DbConn('mydb.db')
dbs = DbSession(my_db)

Standard Query

# Query all users (be careful when there are many records)
users = dbs.select(User).all()

# Query the user with id 1
user = dbs.select(User).filter(User.id == 1).first()

# Query users whose names start with A
users = dbs.select(User).filter(User.name.startswith('A')).all()

# Query user profiles with age >= 25
profiles = dbs.select(UserProfile).filter(UserProfile.age >= 25).all()

# Query articles whose title contains "Python"
arts = dbs.select(Article).filter(Article.title.contains('Python')).all()

# Query articles within a specific range
arts = dbs.select(Article).filter(Article.id.in_([1, 2, 3])).all()

Due to type annotations and chain calls, writing query code is very simple.

If the query ends with all(), it returns a list of model instances; if it ends with first(), it returns a single model instance. If the query result is empty, The former returns an empty list [ ] , while the latter returns None .

Query conditions can use the following comparison operators or functions:

==, !=, <, <=, >, >=
in_(), startswith(), endswith(), contains(), like()

Multi-condition Query

By default, it's an AND query, and the operator can be omitted. For OR queries, you need to use the | operator, and the items at both ends need to be wrapped in parentheses. For combined queries, the AND operator can no longer be omitted, and & must also be written.

# AND query #
# age equals 25, and weight less than 70
dbs.select(UserProfile).filter(UserProfile.age == 25, UserProfile.weight < 70).all()

# OR query #
# age equals 25, or weight less than 70
dbs.select(UserProfile).filter((UserProfile.age == 25) | (UserProfile.weight < 70)).all()

# Combined query #
# all males, or females with age less than 25 (note the parentheses and their operator precedence)
dbs.select(UserProfile).filter((UserProfile.sex == 1) | ((UserProfile.age < 25) & (UserProfile.sex == 0))).all()

Relational Query

For all queries so far, if you print the query results, you'll find some content is missing, for example:

user = dbs.select(User).filter(User.id == 1).first()
print(user)

Assume this user has a profile and has published articles, but the previously defined profile and arts fields are still None. This is because relational queries are not performed by default (because often it's just a single-table query and doesn't need relational data).

To populate relational data, add the preload method to the original query:

user = dbs.select(User).filter(User.id == 1).preload('profile', 'arts').first()
print(user)

Now the profile and arts fields have content.

💡The parameter of preload is the associated field name in string format, with multiple field names separated by commas.

If querying a list, will the N + 1 problem occur?

users = dbs.select(User).filter(User.id.in_([1, 2, 3])).preload('profile', 'arts').all()

The answer is: No. Because pylite-orm automatically optimizes queries, each relational field is queried only once, avoiding the N + 1 problem.

Custom Join Query

The reason why the user instance has profile and arts fields is that when defining the User model, the relational relationship was specified through RelationField. But what if there's no prior definition, yet you need to perform a temporary relational query?

In this case, you can use the join method, along with its target model parameter and relational condition parameter.

Note that the returned content will no longer be model instances or instance lists, but can only be dictionaries or dictionary lists. Because such a data model simply doesn't exist.

arts = dbs.select(Article).join(UserProfile, Article.uid == UserProfile.uid).serial_list()

Note: If you end the query with first() or all() at this point, an error will occur. You must use serial() or serial_list() instead. These two methods return a serialized object or a list of serialized objects (i.e., a dictionary or a list of dictionaries).

💡 The join method has a total of 3 parameters. The first is the associated target model, the second is the association condition, and the third is the join type, which defaults to JoinType.LEFT.

join queries can continue to use filter for filtering.

In many cases, join queries need to specify field names and rename fields (to avoid duplicates), so continue adding columns:

arts = dbs.select(Article).join(UserProfile, Article.uid == UserProfile.uid).columns(
  'article.id, article.title, user_profile.uid AS userid', 'user_profile.age').serial_list()

Value Query

Sometimes we only need to query one or some values, for example, returning a user's name based on id, or returning a set of names based on a set of ids, or returning age and height based on id.

The session instance provides 2 methods: value and values.

value:

result = dbs.select(User).filter(User.id == 1).value('name')
print(result)  ## 'Tom'

values:

result = dbs.select(User).filter(User.id.in_([1, 2, 3])).values('name')
print(result)  ## ['Tom', 'Jerry', 'Alice']

Another usage of values:

result = dbs.select(UserProfile).filter(UserProfile.uid == 1).values('age', 'weight')
print(result)  ## [25, 70]

result = dbs.select(UserProfile).filter(UserProfile.uid.in_([1, 2])).values('age', 'weight')
print(result)  ## [(25, 70), (26, 75)]

Obviously, value queries return strings, numbers, or lists. For multi-field queries with more than 1 row (like the last case above), the returned data will be wrapped in tuples.

If the query result is empty, value returns None and values returns an empty list.

Curious about what values returns when querying a single value?

result = dbs.select(User).filter(User.id == 1).values('name')
print(result)  ## ['Tom']

It returned a list, even though there's only 1 value.

💡When doing value queries, it can't end with all(), first(), or serial(), serial_list(), because they all have clear semantics: returning objects, object lists, dictionaries, or dictionary lists. Value queries return values or value lists.

Native Where Query

pylite-orm only provides support for common queries. For special queries, you can use the where method to directly write native SQLite where statements:

arts = dbs.select(Article).where("article.id = 1").all()

For using native where, there's only one thing to note: if you're concatenating strings and there's content from user input, be sure to check for SQL injection.

Field Functions

pylite-orm currently supports 4 aggregate functions: count, sum, max, min, and 2 time functions: date, strftime, plus a special rename function: as_.

These functions can only be used in columns or value/values.

Here are usage examples:

# Return user info, creation time returns only the date part and is renamed to ymd
result = dbs.select(User).columns('id', 'name', Func.date('create_at').as_('ymd')).serial()
print(result)  ## [{'id': 1, 'name': 'Tom', 'ymd': '2020-01-01'}, ...]

# Group by sex, return the maximum age in each group
result = dbs.select(User).columns('sex', Func.max('age').as_('max_age')).group_by('sex').serial_list()
print(result)  ## [{'sex': 1, 'max_age': 30}, {'sex': 0, 'max_age': 28}]

# Query total number of .users.
result = dbs.select(User).values(Func.count('id'))
print(result)  ## 3

In the example code, to work with the max function, the group_by method appears, which is what we'll introduce next.

Pagination, Sorting, Grouping

Pagination uses limit and offset methods:

# 10 records per page, page 1
result = dbs.select(User).limit(10).offset(0).all()

Sorting uses the order_by method:

# Sort by age ascending
result = dbs.select(User).order_by('age').all()

# Sort by age descending
result = dbs.select(User).order_by('-age').all()

Grouping uses the group_by method:

# Group by sex, return the maximum age in each group
result = dbs.select(User).columns('sex', Func.max('age')).group_by('sex').serial_list()

Return Serialized Data

Through the previous examples, you've become familiar with the serial() method, which converts query results to serialized data, i.e., dictionaries or dictionary lists.

It's worth mentioning that the dictionaries returned by pylite-orm's serial() method support object-style access, with type annotation AttrDict. This means you can access fields in the dictionary just like regular objects:

result = dbs.select(User).columns('id', 'name').filter(User.id == 1).serial()
print(result['name'])  ## 'Tom'
print(result.name)     ## 'Tom'

Both traditional dictionary style and object access style work!

There's another scenario, for example with FastAPI, where you want the returned data to be Pydantic models so they can be output directly to the client. In this case, pass response_model as a parameter to the serial() method to get the expected result:

from pydantic import BaseModel

class UserOut(BaseModel):
    id: int
    name: str

result = dbs.select(User).columns('id', 'name').filter(User.id == 1).serial(UserOut)
print(result)  ## UserOut(id=1, name='Tom')

💡 Pydantic models need to be defined in advance, and field names must match database field names.

In fact, you can arbitrarily define your own data models, as long as they support initialization parameters. For example, a custom dataclass, then pass it to serial, and the data will be returned in the expected structure format.

Maximizes your development flexibility.

Revisiting Chain Calls

A complex query can have very long statements, and it might be dynamically generated based on business logic. pylite-orm's chain calls provide perfect support for this:

age_val = 18
size = 10
page = 1
query = dbs.select(User)
if some_condition:
    query = query.filter(User.age > age_val)
if some_other_condition:
    query = query.limit(size).offset(page)
result = query.all()

Keys	Action
`?`	Open this help
`n`	Next page
`p`	Previous page
`s`	Search