Custom Queries¶

When dynamic queries aren't expressive enough, use the @Query decorator to write raw SQL.

Basic usage¶

from py_spring_model import CrudRepository, Query
from typing import List, Optional

class UserRepository(CrudRepository[int, User]):
    @Query("SELECT * FROM user WHERE age > :min_age")
    def find_users_older_than(self, min_age: int) -> List[User]: ...

    @Query("SELECT * FROM user WHERE email LIKE '%' || :domain || '%'")
    def find_users_by_email_domain(self, domain: str) -> List[User]: ...

The @Query decorator:

Registers the method to be skipped by auto-implementation
At runtime, substitutes parameters into the SQL template using SQLAlchemy's text() bind parameters
Maps the result rows to your model class using Pydantic's model_validate

Parameter substitution¶

Use :parameter_name syntax (SQLAlchemy bind parameters) in your SQL:

@Query("SELECT * FROM user WHERE age BETWEEN :min_age AND :max_age")
def find_users_by_age_range(self, min_age: int, max_age: int) -> List[User]: ...

@Query("SELECT * FROM user WHERE name = :name AND status = :status LIMIT 1")
def get_user_by_name_and_status(self, name: str, status: str) -> Optional[User]: ...

Method parameters are validated against the SQL template at runtime — missing or mistyped parameters raise clear errors.

Return types¶

List of models¶

@Query("SELECT * FROM user WHERE age > :min_age")
def find_older_users(self, min_age: int) -> List[User]: ...

Optional single model¶

@Query("SELECT * FROM user WHERE email = :email LIMIT 1")
def get_user_by_email(self, email: str) -> Optional[User]: ...

Returns None if no row matches.

Scalar types¶

@Query("SELECT COUNT(*) FROM user WHERE status = :status")
def count_by_status(self, status: str) -> int: ...

Supports int, float, str, and bool return types.

None (fire-and-forget)¶

@Query("DELETE FROM user WHERE status = :status", is_modifying=True)
def purge_by_status(self, status: str) -> None: ...

Modifying queries¶

For INSERT, UPDATE, and DELETE operations, set is_modifying=True. This ensures the session commits after execution:

@Query(
    "UPDATE user SET status = :new_status WHERE age < :max_age",
    is_modifying=True,
)
def deactivate_young_users(self, new_status: str, max_age: int) -> List[User]: ...

@Query(
    "DELETE FROM user WHERE status = :status",
    is_modifying=True,
)
def purge_users_by_status(self, status: str) -> List[User]: ...

Warning

Without is_modifying=True, write operations will not be committed to the database.

When to use @Query vs dynamic methods¶

Scenario	Use
Simple equality, comparison, or membership filters	Dynamic methods
`BETWEEN`, null checks, pattern matching	Dynamic methods with field operations
Single-level relationship joins	Dynamic methods with relationship queries
Count, exists, delete by condition	Dynamic methods with `count_by_`, `exists_by_`, `delete_by_`
Multi-level joins, subqueries, aggregations	`@Query`
`ORDER BY`, `LIMIT` in combination	`@Query`
INSERT/UPDATE operations	`@Query` with `is_modifying=True`
Complex `WHERE` clauses	`@Query`

Combining with dynamic methods¶

A repository can mix dynamic methods and @Query methods freely:

class UserRepository(CrudRepository[int, User]):
    # Dynamic — auto-implemented
    def find_by_name(self, name: str) -> Optional[User]: ...
    def find_all_by_status(self, status: str) -> List[User]: ...

    # Custom SQL
    @Query("SELECT * FROM user WHERE age BETWEEN :min AND :max ORDER BY age")
    def find_in_age_range(self, min: int, max: int) -> List[User]: ...

    @Query("UPDATE user SET status = 'archived' WHERE age > :age", is_modifying=True)
    def archive_old_users(self, age: int) -> None: ...