Cheat Sheet
Authentication and Session Handling
Login
Create a Directus instance using email and password
from py_directus import Directus
directus = await Directus("https://example.com", email="user@example.com", password="secret")
Alternatively create a Directus instance using the static token
from py_directus import Directus
directus = await Directus("https://example.com", token="static_token")
Another way is to use the with
statement to automatically logout when the session ends
async with Directus(url, email, password) as directus:
# Manually login
await directus.login()
# Manually start cache
await directus.start_cache()
# do stuff
# OR
async with await Directus(url, email, password) as directus:
# do stuff
Refresh Token
If you want to refresh the token you can use the refresh
method
await directus.refresh()
Logout
Logout from Directus
await directus.logout()
Multiple Users in the Same Session
You can use multiple users in the same session by creating a new Directus instance by passing the client object
connection = httpx.AsyncClient()
directus1 = await Directus(url, token=token, connection=connection)
directus2 = await Directus(url, email=email, password=password, connection=connection)
Collections
There are two ways to set a collection, either by passing the collection name as a string or by passing the collection as a Pydantic model.
Using the collection
method you can pass the collection name as a string
directus.collection("directus_users")
Or you can pass the collection as a Pydantic
model
from typing import Optional
from pydantic import ConfigDict
from py_directus.models import DirectusModel
class User(DirectusModel):
id: Optional[str]
first_name: Optional[str]
last_name: Optional[str]
avatar: Optional[str]
description: Optional[str]
email: Optional[str]
role: Optional[str] | Optional[Role]
status: Optional[str]
title: Optional[str]
token: Optional[str]
model_config = ConfigDict(collection="directus_users")
directus.collection(User)
Don't forget to set the
collection
attribute in themodel_config
attribute
If you go with the second option, you will get the responses as Pydantic
models (auto parsing)
The
collection
method returns aDirectusRequest
object which is used to perform READ, CREATE, UPDATE and DELETE operations
Reading Data
When you have the DirectusRequest
object you can use the read
method to get the data.
This will return a DirectusResponse
object which contains the data.
Imporatnt note: The
read
method must be awaited
await directus.collection("directus_users").read()
Filtering
For an easy equality filter you can pass the field name and the value
await directus.collection("directus_users").filter(first_name="John").read()
To add multiple equality filters you can chain the filter
method
await directus.collection("directus_users")
.filter(first_name="John")
.filter(last_name="Doe").read()
Using it like this you chain the filters with AND
operator
F objects
To define complex logic in filters, use the F
object
from py_directus import F
await directus.collection("directus_users")
.filter(
(F(first_name="John") | F(first_name="Jane"))
& F(last_name="Doe")
).read()
Important note: The
F
object does not support negation
Sorting
You can sort the data by passing the field name to the sort
method
await directus.collection("directus_users").sort("first_name", asc=True).read()
To add multiple sorting fields you can chain the sort
method
await directus.collection("directus_users")
.sort("first_name", asc=True)
.sort("last_name", asc=False).read()
Limiting
You can limit the data by passing the limit to the limit
method
await directus.collection("directus_users").limit(10).read()
Aggregation
Aggregate the number of records in the query
await directus.collection("directus_users").aggregate().read()
# OR
await directus.collection("directus_users").aggregate(count="*").read()
To add multiple aggregates you can chain the aggregate
method
await directus.collection("products")
.aggregate(countDistinct="id")
.aggregate(sum="price").read()
Agg objects
You can aggregate the data by defining the needed aggregation with the Agg
class and passing it to the aggregate
method
from py_directus.aggregator import Agg
agg_obj = Agg(operator=AggregationOperators.Count)
await directus.collection("directus_users").aggregate(agg_obj).read()
In case you need only certain fields
from py_directus.aggregator import Agg
amount_agg = Agg(operator=AggregationOperators.Sum, fields="amount")
await directus.collection("transactions").aggregate(amount_agg).read()
The available aggregation operators are:
- Count
- CountDistinct
- CountAll (Only in GraphQL)
- Sum
- SumDistinct
- Average
- AverageDistinct
- Minimum
- Maximum
Grouping
You can group the data by passing the field names to the group_by
method
await directus.collection("directus_users").group_by("first_name", "last_name").read()
Searching
You can search the data by passing the search term to the search
method
await directus.collection("directus_users").search("John").read()
Selecting Fields
You can select the fields you want to get by passing the field names to the fields
method
await directus.collection("directus_users").fields("first_name", "last_name").read()
Getting the Count Metadata
You can get the count of the data (total count and filtered count) calling include_count
await directus.collection("directus_users").include_count().read()
CRUD
Retrieving items
After you call read()
you get a DirectusResponse
object which contains the data.
item
for single itemitems
for multiple items
Getting the data as a dictionary or a list of dictionaries
response = await directus.collection("directus_users").read()
print(response.item["first_name"])
print(response.items)
If you provide the collection
method a Pydantic
model you will get the data as a Pydantic
object or a list of Pydantic
objects
response = await directus.collection(User).read()
print(response.item.first_name)
print(response.items)
Converting to Models (pydantic) or to Dictionary
Apart from the auto parsing, you can manually convert the data to a Pydantic
model instance or to a dictionary using:
item_as(User)
oritems_as(User)
item_as_dict()
oritems_as_dict()
response = await directus.collection("directus_users").read()
print(response.item_as(User))
response = await directus.collection(User).read()
print(response.item_as_dict())
Creating Items
The library does not support Pydantic
models for creation, you have to pass a dictionary
- create(items: dict|List[dict])
await directus.collection("directus_users").create({
"first_name": "John", "last_name": "Doe"
})
# OR
await directus.collection("directus_users").create(
[
{"first_name": "John", "last_name": "Doe"},
{"first_name": "Jane", "last_name": "Doe"}
]
)
Updating Items
The library do not support Pydantic
models for updating, you have to pass a dictionary
update(ids: str|int, items: dict)
update(ids: List[str|int], items: List[dict])
await directus.collection("directus_users").update(1, {
"first_name": "Red",
"last_name": "John"
})
# OR
await directus.collection("directus_users").update(
[1, 2],
[
{"first_name": "Jean-Luc"},
{"first_name": "Jane", "last_name": "Doe"}
]
)
Deleting Items
delete(ids: str|int|List[str|int])
await directus.collection("directus_users").delete(1)
# OR
await directus.collection("directus_users").delete([1, 2])
Supporting
Pydantic
models forcreate
/update
/delete
item operations is shortly coming.