Record Queries

This is the v1.0 documentation, for a release that is still in development. The API it describes is not on PyPI and can still change.

Reading about the version you have installed? Switch to v0.x.

Internal engineering reference for the Sillo ORM query layer: scopes, casts, query helpers, pagination, and query logging.

Source: core/sillo/record/scopes.py, core/sillo/record/casting.py, core/sillo/record/queries.py, core/sillo/record/pagination.py, core/sillo/record/logging.py

---

1. Overview

The query layer sits between the developer’s fluent API and Tortoise ORM’s query builder. It adds three capabilities that Tortoise does not provide:

1. Scopes: reusable query fragments that can be chained (local) or applied automatically (global).
2. Casts: transparent encode/decode of field values at the attribute boundary, so the database always sees the serialized form and the developer always sees the Python-native form.
3. Query helpers: pagination, async iteration, explain plans, bulk lookups, and in-memory aggregation.

graph TB
    subgraph "Developer API"
        A["User.active().vip().paginate()"]
    end
    subgraph "Scope Layer"
        B[RecordQuerySet]
        C[ScopeRegistry]
        D[RecordManager]
    end
    subgraph "Cast Layer"
        E[HasCasts]
        F[CastRegistry]
    end
    subgraph "Query Helpers"
        G[paginate]
        H[iter_all]
        I[explain]
        J[find_by_ids]
        K[count_by]
    end
    subgraph "Tortoise ORM"
        L[QuerySet]
        M[Database]
    end

    A --> B
    B -->|scope interception| C
    D -->|get_queryset| B
    D -->|apply_scopes| C
    B --> L
    L --> M
    E -->|cast_get/cast_set| F
    G --> L
    H --> L
    I --> L
    J --> L
    K --> L

---

2. ScopeRegistry

File: core/sillo/record/scopes.py

class ScopeRegistry:
    def __init__(self):
        self._global_scopes: list[Callable] = []

    def add(self, scope: Callable) -> None:
        self._global_scopes.append(scope)

    def remove(self, scope: Callable) -> bool:
        try:
            self._global_scopes.remove(scope)
            return True
        except ValueError:
            return False

    def apply(self, queryset):
        for scope in self._global_scopes:
            queryset = scope(queryset)
        return queryset

    def without_global_scopes(self, queryset):
        return queryset

2.1 Design

• Global scopes are callables with the signature (queryset) -> queryset.
• They are stored in a list and applied in registration order.
• apply() chains them: each scope receives the queryset returned by the previous one.
• remove() returns True if found, False otherwise. It uses list.remove which matches by identity (is), not equality. The exact same callable object must be passed.
• without_global_scopes() is a pass-through that returns the queryset unchanged. It exists as a named method so that HasScopes.without_global_scopes() can delegate to it with a clear intent.

2.2 Lifecycle

Each model class gets its own ScopeRegistry instance, stored on cls._scope_registry. The registry is created lazily on the first call to add_global_scope:

@classmethod
def add_global_scope(cls, scope: Callable) -> None:
    if cls._scope_registry is None:
        cls._scope_registry = ScopeRegistry()
    cls._scope_registry.add(scope)

This means models that never register a global scope pay no cost.

---

3. HasScopes

File: core/sillo/record/scopes.py

class HasScopes:
    _scope_registry: ScopeRegistry | None = None

    @classmethod
    def add_global_scope(cls, scope: Callable) -> None:
        if cls._scope_registry is None:
            cls._scope_registry = ScopeRegistry()
        cls._scope_registry.add(scope)

    @classmethod
    def without_global_scopes(cls):
        return cls._meta.manager.without_global_scopes()

    @classmethod
    def apply_scopes(cls, queryset):
        if cls._scope_registry is not None:
            return cls._scope_registry.apply(queryset)
        return queryset

3.1 Local Scopes

Local scopes are methods prefixed with scope_. They are defined on the model class and receive the queryset as the first argument:

class User(Model):
    @classmethod
    def scope_active(cls, queryset):
        return queryset.filter(is_active=True)

    @classmethod
    def scope_vip(cls, queryset, plan_type="vip"):
        return queryset.filter(plan=plan_type)

The __init_subclass__ hook on Model (see doc 21, §4) auto-generates shortcut classmethods:

User.active()        # → User.all().active()
User.vip("premium")  # → User.all().vip("premium")

3.2 Global Scopes

Global scopes are applied to every query on the model:

def tenant_scope(queryset):
    return queryset.filter(tenant_id=get_current_tenant())

Tenant.add_global_scope(tenant_scope)

# Every query now includes WHERE tenant_id = ?
users = await User.all()           # filtered
users = await User.filter(...)     # filtered
users = await User.without_global_scopes().all()  # unfiltered

3.3 Scope Interception via RecordQuerySet

Local scopes are not methods on the queryset. They are methods on the model that receive the queryset. The RecordQuerySet.__getattr__ bridge makes them chainable:

class RecordQuerySet(QuerySet):
    def __getattr__(self, name: str):
        scope = getattr(self.model, f"scope_{name}", None)
        if scope is None:
            raise AttributeError(...)

        def apply_local_scope(*args, **kwargs):
            return scope(self, *args, **kwargs)

        return apply_local_scope

When the developer writes User.active().vip(), the chain resolves as:

1. User.active() → User.all().active() (via __init_subclass__ shortcut)
2. User.all() returns a RecordQuerySet (via RecordManager.get_queryset)
3. .active() → RecordQuerySet.__getattr__("active") → calls User.scope_active(self) → returns filtered queryset
4. .vip() → same mechanism on the new queryset

sequenceDiagram
    participant Dev as Developer
    participant InitSub as __init_subclass__
    participant QS as RecordQuerySet
    participant Model as User model
    participant Scope as scope_active

    Dev->>InitSub: User.active()
    InitSub->>QS: User.all() → RecordQuerySet
    InitSub->>QS: .active()
    QS->>QS: __getattr__("active")
    QS->>Model: getattr(User, "scope_active")
    Model-->>QS: scope_active function
    QS->>Scope: scope_active(self, *args)
    Scope->>QS: queryset.filter(is_active=True)
    QS-->>Dev: filtered RecordQuerySet

3.4 without_global_scopes

# On RecordQuerySet
def without_global_scopes(self):
    queryset = self.__class__(self.model)
    queryset._db = self._db
    return queryset

# On RecordManager
def without_global_scopes(self) -> RecordQuerySet:
    return RecordQuerySet(self._model)

Creates a fresh RecordQuerySet that bypasses the manager’s get_queryset() (which applies global scopes). The database connection is preserved from the original queryset.

---

4. RecordManager

File: core/sillo/record/scopes.py

class RecordManager(Manager):
    def get_queryset(self) -> RecordQuerySet:
        queryset = RecordQuerySet(self._model)
        apply_scopes = getattr(self._model, "apply_scopes", None)
        if apply_scopes is not None:
            return apply_scopes(queryset)
        return queryset

    def without_global_scopes(self) -> RecordQuerySet:
        return RecordQuerySet(self._model)

4.1 How Tortoise Uses the Manager

When you call User.all(), Tortoise’s Model.all() delegates to cls._meta.manager.get_queryset(). Sillo’s RecordManager overrides this to:

1. Create a RecordQuerySet (instead of Tortoise’s default QuerySet).
2. Apply global scopes via apply_scopes().

The manager is assigned in __init_subclass__:

def __init_subclass__(cls, **kwargs):
    super().__init_subclass__(**kwargs)
    if hasattr(cls, "_meta"):
        cls._meta.manager = RecordManager(cls)

4.2 Manager Diagram

graph LR
    A["User.all()"] --> B["Model.all()"]
    B --> C["_meta.manager.get_queryset()"]
    C --> D["RecordManager.get_queryset()"]
    D --> E["RecordQuerySet(model)"]
    E --> F["apply_scopes(queryset)"]
    F --> G["ScopeRegistry.apply()"]
    G --> H["Filtered RecordQuerySet"]

---

5. HasCasts: Attribute Casting

File: core/sillo/record/casting.py

5.1 CastRegistry

class CastRegistry:
    _builtins: ClassVar[dict[str, tuple[Callable, Callable]]] = {}

    @classmethod
    def register(cls, name: str, encoder: Callable, decoder: Callable) -> None:
        cls._builtins[name] = (encoder, decoder)

    @classmethod
    def get(cls, name: str):
        return cls._builtins.get(name)

Each cast is a pair of (encoder, decoder):

• Encoder: called on cast_set (before save). Transforms Python values to database-safe representations.
• Decoder: called on cast_get (after read). Transforms database values back to Python-native types.

5.2 Built-in Casts

Name	Encoder	Decoder
"json"	json.dumps(value, default=str)	json.loads(value)
"datetime"	value.isoformat()	datetime.fromisoformat(value)
"bool"	bool(v)	bool(v)
"int"	int(v)	int(v) if not None
"float"	float(v)	float(v) if not None

Registered at module load time:

CastRegistry.register("json", _json_encoder, _json_decoder)
CastRegistry.register("datetime", _datetime_encoder, _datetime_decoder)
CastRegistry.register("bool", lambda v: bool(v), lambda v: bool(v))
CastRegistry.register("int", lambda v: int(v), lambda v: int(v) if v is not None else None)
CastRegistry.register("float", lambda v: float(v), lambda v: float(v) if v is not None else None)

5.3 Encrypted Cast

The "encrypted" cast is special. It requires a key and is created via a factory:

def _encrypted_factory(key: str):
    def encoder(value: str) -> str:
        encoded = bytes([ord(c) ^ ord(key[i % len(key)]) for i, c in enumerate(value)])
        return base64.b64encode(encoded).decode()

    def decoder(value: str) -> str:
        decoded = base64.b64decode(value)
        return "".join(chr(b ^ ord(key[i % len(key)])) for i, b in enumerate(decoded))

    return encoder, decoder

Usage in _casts:

class User(Model):
    _casts = {
        "secret_key": ("encrypted", {"key": "my-secret"}),
    }

The tuple form ("encrypted", {"key": "..."}) is detected by get_cast:

def get_cast(self, field_name: str):
    cast_def = self._casts.get(field_name)
    if cast_def is None:
        return None, None
    if isinstance(cast_def, str):
        return CastRegistry.get(cast_def) or (None, None)
    if isinstance(cast_def, tuple):
        name, kwargs = cast_def[0], cast_def[1] if len(cast_def) > 1 else {}
        if name == "encrypted":
            return _encrypted_factory(**kwargs)
    if callable(cast_def):
        return cast_def()
    return None, None

5.4 HasCasts Mixin

class HasCasts:
    _casts: ClassVar[dict[str, Any]] = {}

    def get_cast(self, field_name: str):
        ...  # resolves encoder/decoder from _casts

    def cast_get(self, field_name: str, value: Any) -> Any:
        _, decoder = self.get_cast(field_name)
        if decoder and value is not None:
            return decoder(value)
        return value

    def cast_set(self, field_name: str, value: Any) -> Any:
        encoder, _ = self.get_cast(field_name)
        if encoder and value is not None:
            return encoder(value)
        return value

5.5 Cast Flow Diagram

sequenceDiagram
    participant Dev as Developer
    participant GetAttr as __getattribute__
    participant Casts as HasCasts
    participant DB as Database

    Note over Dev,DB: READ PATH
    Dev->>GetAttr: user.metadata
    GetAttr->>Casts: cast_get("metadata", raw_str)
    Casts->>Casts: get_cast("metadata") → json decoder
    Casts-->>GetAttr: json.loads(raw_str) → dict
    GetAttr-->>Dev: {"key": "value"}

    Note over Dev,DB: WRITE PATH (save)
    Dev->>Dev: user.save()
    Dev->>Casts: _encoded_cast_values context
    Casts->>Casts: cast_set("metadata", dict)
    Casts->>Casts: get_cast("metadata") → json encoder
    Casts-->>DB: json.dumps(dict) → '{"key": "value"}'
    DB-->>Dev: row saved

5.6 Cast Resolution Order

When get_cast(field_name) is called, the resolution depends on the type of _casts[field_name]:

Type of value	Resolution
str (e.g. "json")	Look up in CastRegistry._builtins
tuple	First element is the name; second is kwargs.
	"encrypted" → _encrypted_factory(**kwargs)
callable	Call it; expect (encoder, decoder) return
None / missing	No cast; pass-through

---

6. Query Helpers

File: core/sillo/record/queries.py

6.1 paginate

async def paginate(queryset, page=1, page_size=20, *, ordering=None) -> PaginatedResult:
    if ordering:
        queryset = queryset.order_by(ordering)
    total = await queryset.count()
    offset = (page - 1) * page_size
    items = await queryset.offset(offset).limit(page_size).all()
    return PaginatedResult(items=items, total=total, page=page, page_size=page_size)

Key details:

• page is 1-based (not 0-based).
• ordering accepts field names with optional - prefix for descending (e.g., "-created_at").
• Two queries are executed: COUNT(*) and SELECT ... OFFSET ... LIMIT.
• Returns a PaginatedResult with metadata.

6.2 PaginatedResult

class PaginatedResult:
    def __init__(self, items, total, page, page_size):
        self.items = items
        self.total = total
        self.page = page
        self.page_size = page_size

    @property
    def pages(self) -> int:
        return max(1, (self.total + self.page_size - 1) // self.page_size)

    @property
    def has_next(self) -> bool:
        return self.page < self.pages

    @property
    def has_prev(self) -> bool:
        return self.page > 1

    def to_dict(self) -> dict:
        return {
            "items": [item.to_dict() if hasattr(item, "to_dict") else str(item)
                      for item in self.items],
            "total": self.total,
            "page": self.page,
            "page_size": self.page_size,
            "pages": self.pages,
            "has_next": self.has_next,
            "has_prev": self.has_prev,
        }

Properties:

• pages: total number of pages (ceiling division, minimum 1).
• has_next / has_prev: boolean navigation hints.
• to_dict(): serializes the entire result for JSON responses.

6.3 iter_all

async def iter_all(queryset, batch_size=500) -> AsyncIterator[Any]:
    offset = 0
    while True:
        batch = await queryset.offset(offset).limit(batch_size).all()
        if not batch:
            break
        for item in batch:
            yield item
        offset += batch_size

• Memory-efficient async generator for large datasets.
• Fetches batch_size rows at a time using OFFSET/LIMIT.
• Yields individual items, not batches: the caller sees a flat stream.
• Caveat: OFFSET-based pagination is inefficient for very large datasets (the DB must scan and skip rows). For millions of rows, consider cursor-based pagination.

6.4 explain

async def explain(queryset) -> str:
    try:
        sql, params = queryset.sql()
        conn = connections.get("default")
        result = await conn.execute_query(f"EXPLAIN {sql}", params)
        return str(result)
    except Exception as e:
        return f"EXPLAIN unavailable: {e}"

• Extracts the SQL from a queryset via .sql().
• Prepends EXPLAIN and executes against the default connection.
• Returns the execution plan as a string.
• Catches all exceptions: EXPLAIN support varies by backend.

6.5 find_by_ids

async def find_by_ids(queryset, ids: list[Any]) -> list[Any]:
    pk = queryset.model._meta.pk_attr
    return await queryset.filter(**{f"{pk}__in": ids}).all()

• Fetches multiple rows by primary key in a single query.
• Uses Tortoise’s __in filter operator.
• The primary key field is resolved dynamically from _meta.pk_attr.

6.6 count_by

async def count_by(queryset, field: str) -> dict:
    results = {}
    async for row in queryset.all():
        val = getattr(row, field, None)
        results[str(val)] = results.get(str(val), 0) + 1
    return results

• In-memory group-by aggregation.
• Iterates all rows and counts occurrences of each field value.
• Values are stringified for dict keys.
• Performance warning: This loads every row into memory. For large datasets, use a database-level GROUP BY via raw SQL or Tortoise’s .annotate().

---

7. TortoiseDataHandler

File: core/sillo/record/pagination.py

Bridges Sillo’s pagination strategies (PageNumberPagination, LimitOffsetPagination, CursorPagination) to Tortoise querysets.

7.1 AsyncTortoiseDataHandler

class TortoiseDataHandler(AsyncDataHandler):
    def __init__(self, queryset):
        self._qs = queryset

    async def get_total_items(self) -> int:
        return await self._qs.count()

    async def get_items(self, offset: int, limit: int) -> list[Any]:
        return await self._qs.offset(offset).limit(limit).all()

Implements the AsyncDataHandler protocol from sillo.pagination:

class AsyncDataHandler(Protocol):
    async def get_total_items(self) -> int: ...
    async def get_items(self, offset: int, limit: int) -> list[Any]: ...

7.2 SyncTortoiseDataHandler

class SyncTortoiseDataHandler(SyncDataHandler):
    def __init__(self, data: list[Any]):
        self._data = data

    def get_total_items(self) -> int:
        return len(self._data)

    def get_items(self, offset: int, limit: int) -> list[Any]:
        return self._data[offset : offset + limit]

For pre-fetched data that needs synchronous pagination (rare, mostly for compatibility with SyncPaginator).

7.3 Usage with Pagination Strategies

from sillo.record.pagination import TortoiseDataHandler
from sillo.pagination import PageNumberPagination

handler = TortoiseDataHandler(User.filter(is_active=True))
paginator = PageNumberPagination()
result = await paginator.paginate(handler, page=2, page_size=10)

---

8. QueryLogger

File: core/sillo/record/logging.py

Tracks database queries issued during a request and detects slow queries and N+1 patterns.

8.1 QueryLogEntry

class QueryLogEntry:
    def __init__(self, sql: str, params: Any, duration_ms: float, source: str = ""):
        self.sql = sql
        self.params = params
        self.duration_ms = duration_ms
        self.source = source
        self.timestamp = time.time()

Each entry captures the SQL, parameters, duration, optional source annotation, and wall-clock timestamp.

8.2 QueryLogger

class QueryLogger:
    def __init__(self, slow_threshold_ms=100.0, detect_n_plus_one=True):
        self._entries: list[QueryLogEntry] = []
        self._slow_threshold = slow_threshold_ms
        self._detect_n1 = detect_n_plus_one
        self._started = False
        self._start_time = 0.0

Lifecycle:

1. start(): clears entries, sets _started = True, records start time.
2. log(sql, params, duration_ms, source): appends an entry. If duration_ms > slow_threshold, logs a warning.
3. stop(): sets _started = False.
4. report(): returns a summary dict.

8.3 Slow Query Detection

def log(self, sql, params=None, duration_ms=0, source="") -> None:
    if self._started:
        entry = QueryLogEntry(sql, params, duration_ms, source)
        self._entries.append(entry)
        if duration_ms > self._slow_threshold:
            logger.warning("SLOW QUERY [%.1fms] %s", duration_ms, sql[:200])

• Default threshold: 100ms.
• Slow queries are logged at WARNING level to sillo.record.logging.
• The SQL is truncated to 200 characters in the log message.

8.4 N+1 Detection

def detect_n_plus_one(self) -> list[str]:
    warnings = []
    sql_list = [e.sql for e in self._entries]
    for i, sql in enumerate(sql_list):
        count = sql_list.count(sql)
        if count > 5:
            warnings.append(f"N+1 detected: query '{sql[:100]}' ran {count} times")
    return list(set(warnings))

Algorithm:

1. Collect all SQL strings from entries.
2. For each unique SQL, count how many times it appears.
3. If a query ran more than 5 times, flag it as a potential N+1.
4. Deduplicate warnings with set().

Limitations:

• Parameterized queries with different params are counted separately (correct).
• But SELECT * FROM posts WHERE user_id = ? with different ? values will have the same SQL string, so they are flagged as N+1, which is the intended behavior.
• The threshold of 5 is hardcoded. A configurable threshold would be better.

8.5 Report

def report(self) -> dict[str, Any]:
    return {
        "total_queries": self.total_queries,
        "total_time_ms": self.total_time_ms,
        "slow_queries": len(self.slow_queries),
        "slow_details": [str(e) for e in self.slow_queries],
        "n_plus_one_warnings": self.detect_n_plus_one() if self._detect_n1 else [],
    }

Example output:

{
    "total_queries": 47,
    "total_time_ms": 234.5,
    "slow_queries": 3,
    "slow_details": [
        "[152.3ms] SELECT * FROM posts WHERE user_id = ...",
        "[201.1ms] SELECT COUNT(*) FROM comments WHERE ...",
        "[110.8ms] SELECT * FROM tags WHERE post_id IN ..."
    ],
    "n_plus_one_warnings": [
        "N+1 detected: query 'SELECT * FROM posts WHERE user_id = ...' ran 12 times"
    ]
}

8.6 Usage Pattern

from sillo.record.logging import QueryLogger

logger = QueryLogger(slow_threshold_ms=50)
logger.start()

# ... process ctx ...

logger.stop()
report = logger.report()

if report["n_plus_one_warnings"]:
    for warning in report["n_plus_one_warnings"]:
        print(f"⚠ {warning}")

if report["slow_queries"] > 0:
    print(f"⚠ {report['slow_queries']} slow queries detected")

8.7 Properties

Property	Type	Description
total_time_ms	float	Sum of all query durations
total_queries	int	Number of logged queries
slow_queries	list[QueryLogEntry]	Entries exceeding the threshold

---

9. Integration: How Queries Flow Through the Stack

graph TB
    subgraph "Request Handler"
        A["await User.active().vip().all()"]
    end

    subgraph "RecordManager"
        B["get_queryset()"]
        C["RecordQuerySet(User)"]
        D["apply_scopes(queryset)"]
    end

    subgraph "ScopeRegistry"
        E["tenant_scope(queryset)"]
    end

    subgraph "RecordQuerySet"
        F["__getattr__('active')"]
        G["User.scope_active(self)"]
        H["__getattr__('vip')"]
        I["User.scope_vip(self)"]
    end

    subgraph "Tortoise QuerySet"
        J[".all()"]
        K["SQL: SELECT * FROM users WHERE is_active=true AND plan='vip' AND tenant_id=?"]
    end

    subgraph "HasCasts (on result)"
        L["__getattribute__ for each field"]
        M["cast_get if _casts defined"]
    end

    A --> B
    B --> C
    C --> D
    D --> E
    E --> F
    F --> G
    G --> H
    H --> I
    I --> J
    J --> K
    K --> L
    L --> M

---

10. Pagination Strategies

Sillo’s pagination system (in sillo.pagination) is strategy-based. The TortoiseDataHandler adapts Tortoise querysets to the strategy interface.

10.1 PageNumberPagination

Traditional page-based pagination:

• Input: page, page_size
• SQL: COUNT(*) + SELECT ... OFFSET (page-1)*page_size LIMIT page_size

10.2 LimitOffsetPagination

Offset-based pagination:

• Input: offset, limit
• SQL: COUNT(*) + SELECT ... OFFSET offset LIMIT limit

10.3 CursorPagination

Cursor-based pagination (for infinite scroll):

• Input: cursor (opaque), page_size
• SQL: SELECT ... WHERE id > cursor_id LIMIT page_size
• No COUNT(*) query needed: has_next is determined by fetching page_size + 1 items.

---

11. Advanced Patterns

11.1 Composing Scopes

Scopes compose naturally:

class Post(Model):
    @classmethod
    def scope_published(cls, qs):
        return qs.filter(published_at__isnull=False)

    @classmethod
    def scope_recent(cls, qs, days=7):
        cutoff = datetime.now(timezone.utc) - timedelta(days=days)
        return qs.filter(published_at__gte=cutoff)

    @classmethod
    def scope_by_author(cls, qs, author_id):
        return qs.filter(author_id=author_id)

# Usage:
posts = await Post.published().recent(days=30).by_author(user.id).all()

11.2 Global Scope for Multi-Tenancy

def tenant_scope(queryset):
    from sillo.context import get_current_tenant
    return queryset.filter(tenant_id=get_current_tenant())

class TenantModel(Model):
    class Meta:
        abstract = True

    @classmethod
    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        cls.add_global_scope(tenant_scope)

11.3 Combining Casts with Scopes

Casts and scopes are independent. A field can have both:

class Event(Model):
    _casts = {"metadata": "json", "starts_at": "datetime"}

    @classmethod
    def scope_upcoming(cls, qs):
        # starts_at is decoded to datetime by cast_get, but the filter
        # operates on the DB column, which stores ISO strings
        return qs.filter(starts_at__gte=datetime.now(timezone.utc).isoformat())

11.4 Custom Cast Types

Register project-specific casts:

from sillo.record.casting import CastRegistry

def _point_encoder(value: tuple) -> str:
    return f"{value[0]},{value[1]}"

def _point_decoder(value: str) -> tuple:
    x, y = value.split(",")
    return (float(x), float(y))

CastRegistry.register("point", _point_encoder, _point_decoder)

class Location(Model):
    _casts = {"coordinates": "point"}

loc = Location(coordinates=(37.7749, -122.4194))
await loc.save()  # stores "37.7749,-122.4194"
print(loc.coordinates)  # (37.7749, -122.4194)

---

12. Performance Considerations

12.1 N+1 Prevention

The QueryLogger.detect_n_plus_one() method detects N+1 patterns retroactively. For prevention:

1. Use Tortoise’s prefetch_related() or select_related().
2. Use iter_all() for batch processing instead of loading everything at once.
3. Monitor the query count in development with QueryLogger.

12.2 Pagination Performance

• paginate() executes two queries: COUNT(*) and SELECT ... OFFSET/LIMIT.
• For very large tables, COUNT(*) can be slow. Consider caching the count or using cursor pagination (which skips the count).
• OFFSET-based pagination degrades at high page numbers. Use cursor-based pagination for infinite scroll.

12.3 Cast Overhead

• Each cast_get / cast_set call is a function call + the encoder/decoder itself. For JSON fields, this is json.loads / json.dumps per access.
• The _encoded_cast_values context manager iterates all cast fields on every save(). For models with many cast fields, this adds overhead.
• Mitigation: only use casts for fields that genuinely need transformation. Simple string/integer fields should not have casts.

12.4 Scope Application

• Global scopes are applied on every query, including count(), exists(), and filter() chains.
• If a global scope is expensive (e.g., a subquery), it will be executed repeatedly.
• Mitigation: keep global scopes simple (single filter clauses). Move complex logic to local scopes.

---

13. Source File Reference

File	Contents
core/sillo/record/scopes.py	ScopeRegistry, HasScopes, RecordQuerySet, RecordManager
core/sillo/record/casting.py	CastRegistry, built-in casters, _encrypted_factory, HasCasts
core/sillo/record/queries.py	paginate, PaginatedResult, iter_all, explain, find_by_ids, count_by
core/sillo/record/pagination.py	TortoiseDataHandler, SyncTortoiseDataHandler
core/sillo/record/logging.py	QueryLogEntry, QueryLogger

---

14. Gotchas and Known Issues

1. count_by is in-memory: It loads every row and counts in Python. For large tables, use raw SQL GROUP BY or Tortoise’s .annotate().

2. iter_all uses OFFSET: For tables with millions of rows, OFFSET-based pagination becomes slow. Consider keyset pagination (WHERE id > last_id).

3. N+1 detection threshold: The hardcoded threshold of 5 may be too low for some legitimate patterns (e.g. fetching related data for 10 items). Consider making it configurable.

4. Cast registry is global. CastRegistry._builtins is a class variable, shared across all models. Registering a cast with the same name as an existing one overwrites it silently.

5. Scope identity matching. ScopeRegistry.remove() uses list.remove, which matches by identity (is), not equality. A lambda or function defined inline cannot be removed by defining an identical one.

6. without_global_scopes creates a new queryset: The returned queryset has no filters, ordering, or limits from the original. Chain it carefully:

   User.filter(is_active=True).without_global_scopes()  # loses is_active filter!
   User.without_global_scopes().filter(is_active=True)  # correct

7. explain varies by backend: SQLite’s EXPLAIN output is very different from Postgres’s. The function returns whatever the backend produces as a string.