Skip to content

Adapters

Adapters translate between a data source (SQLModel, SQLAlchemy, or custom) and HyperAdmin's view layer.

Built-in Adapters

Adapter Module Works with
SQLModelAdapter hyperadmin.adapters.sqlmodel SQLModel async models
SQLAlchemyAdapter hyperadmin.adapters.sqlalchemy SQLAlchemy async models

HyperAdmin auto-selects the right adapter at registration time via the adapter registry.

Writing a custom adapter

Subclass BaseAdapter and implement all abstract methods:

from hyperadmin.core.adapters import BaseAdapter

class MyAdapter(BaseAdapter):
    async def get(self, pk):
        ...

    async def list(self, page=1, page_size=10, search=None, filters=None, order_by=None):
        # Return (items, total_count)
        ...

    async def create(self, data):
        ...

    async def update(self, pk, data):
        ...

    async def delete(self, pk):
        ...

    async def get_related(self, pk, field):
        ...

    async def get_schema(self):
        ...

Register it with the adapter registry so HyperAdmin can discover it automatically:

from hyperadmin.adapters.registry import adapter_registry
adapter_registry.register(MyModel, MyAdapter)

API Reference

hyperadmin.core.adapters.BaseAdapter

Bases: ABC

Abstract base class for data adapters.

Defines the contract for all data operations (get, list, create, update, delete). Subclass this to add support for a new ORM or data source.

Example 
class MyAdapter(BaseAdapter):
    async def get(self, pk): ...
Source code in src/hyperadmin/core/adapters.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
class BaseAdapter(ABC):
    """Abstract base class for data adapters.

    Defines the contract for all data operations (get, list, create, update, delete).
    Subclass this to add support for a new ORM or data source.

    Example:
        ```python
        class MyAdapter(BaseAdapter):
            async def get(self, pk): ...
        ```
    """

    model: Any

    def __init__(self, model: Any, engine: Any) -> None:
        self.model = model
        self.engine = engine
        self._queryset_filter: QuerysetFilter | None = None

    def get_queryset(self, request: Request | None = None) -> dict[str, Any]:
        """Return additional equality filters merged into ``list()`` and ``get()`` queries.

        The returned mapping has the shape ``{column_name: value}`` and is applied as
        equality predicates **before** any view-layer filters. Use this hook to implement
        features like row-level security or tenant scoping without leaking those concerns
        into the view layer.

        Subclasses may override this method directly, or callers may register a
        per-request callable via :meth:`set_queryset_filter`. The callable is
        re-evaluated on every ``list()`` / ``get()`` — no caching — so it must be pure
        or request-scoped.

        Default behaviour returns an empty dict, which is a no-op (backward compatible).

        Args:
            request: The active Starlette/FastAPI request, when available. ``None`` is
                permitted so the hook is callable from contexts without an HTTP request
                (e.g. management scripts, tests).

        Returns:
            A dict of column-name to value pairs to be ANDed into the query's WHERE
            clause. Returns ``{}`` by default.
        """
        if self._queryset_filter is None:
            return {}
        return self._queryset_filter(request)

    def set_queryset_filter(self, filter_fn: QuerysetFilter) -> None:
        """Register a per-request callable that returns additional queryset filters.

        The callable receives the current :class:`starlette.requests.Request` (or
        ``None``) and must return a dict of ``{column_name: value}`` equality filters.
        It is re-invoked on every ``list()`` / ``get()`` so it may consult
        request-scoped state (e.g. ``request.state.user``).

        Args:
            filter_fn: A callable taking an optional :class:`Request` and returning a
                dict of equality filters.
        """
        self._queryset_filter = filter_fn

    def _resolve_queryset_filters(self, request: Request | None = None) -> dict[str, Any]:
        """Invoke :meth:`get_queryset` and validate the return type.

        Raises:
            TypeError: If ``get_queryset`` returns a value that is not a ``dict``.
        """
        result = self.get_queryset(request)
        if not isinstance(result, dict):
            raise TypeError(
                f"{type(self).__name__}.get_queryset() must return a dict, "
                f"got {type(result).__name__}"
            )
        return result

    @abstractmethod
    async def get(self, pk: Any) -> Any:
        """
        Retrieves a single object by its primary key.

        Args:
            pk: The primary key of the object to retrieve.

        Returns:
            The retrieved object, or None if not found.
        """
        raise NotImplementedError

    @abstractmethod
    async def list(
        self,
        page: int = 1,
        page_size: int = 10,
        search: str | None = None,
        filters: dict[str, Any] | None = None,
        order_by: str | None = None,
        search_fields: builtins.list[str] | None = None,
    ) -> tuple[builtins.list[Any], int]:
        """
        Retrieves a list of objects with optional pagination, searching, and filtering.

        Args:
            page: The page number for pagination.
            page_size: The number of items per page.
            search: A search query to filter the results.
            filters: A dictionary of filters to apply to the query.
            order_by: The field to order the results by.

        Returns:
            A tuple containing the list of objects and the total count of objects.
        """
        raise NotImplementedError

    @abstractmethod
    async def create(self, data: dict[str, Any]) -> Any:
        """
        Creates a new object.

        Args:
            data: A dictionary of data for the new object.

        Returns:
            The created object.
        """
        raise NotImplementedError

    @abstractmethod
    async def update(self, pk: Any, data: dict[str, Any]) -> Any:
        """
        Updates an existing object.

        Args:
            pk: The primary key of the object to update.
            data: A dictionary of data to update the object with.

        Returns:
            The updated object.
        """
        raise NotImplementedError

    @abstractmethod
    async def delete(self, pk: Any) -> None:
        """
        Deletes an object by its primary key.

        Args:
            pk: The primary key of the object to delete.
        """
        raise NotImplementedError

    @abstractmethod
    async def get_related(self, pk: Any, field: str) -> builtins.list[Any]:
        """
        Retrieves related objects for a given object and field.

        Args:
            pk: The primary key of the object.
            field: The name of the relationship field.

        Returns:
            A list of related objects.
        """
        raise NotImplementedError

    @abstractmethod
    async def get_schema(self) -> dict[str, Any]:
        """
        Retrieves the schema definition for the model.

        Returns:
            A dictionary representing the model's schema.
        """
        raise NotImplementedError

    @abstractmethod
    async def get_choices(
        self,
        field: str,
        q: str = "",
        limit: int = 50,
        offset: int = 0,
        **filters: Any,
    ) -> builtins.list[ChoiceItem]:
        """Return paginated, searchable choices for a relation field.

        Args:
            field: The relation field name on this adapter's model.
            q: Optional search string (ILIKE match on string columns).
            limit: Max results to return. Must not exceed 200.
            offset: Number of rows to skip for pagination.
            **filters: Extra equality filters forwarded to the query (cascading support).

        Returns:
            A list of ``ChoiceItem`` dicts with ``value``, ``label``, ``selected=False``.

        Raises:
            ValueError: When ``limit`` exceeds 200.
        """
        raise NotImplementedError

    @abstractmethod
    async def save_inline_rows(
        self,
        spec: InlineModelSpec,
        rows: builtins.list[dict[str, Any]],
        parent_pk: Any,
    ) -> None:
        """Persist validated inline rows — create, update, or delete as needed.

        Args:
            spec: The ``InlineModelSpec`` describing the related model and FK field.
            rows: Validated row dicts, each optionally containing ``_pk`` (for
                update/delete) and ``_delete`` (for deletion).
            parent_pk: The primary key of the parent object to associate new rows with.
        """
        raise NotImplementedError

create(data) abstractmethod async

Creates a new object.

Parameters:

Name Type Description Default
data dict[str, Any]

A dictionary of data for the new object.

 required

Returns:

Type Description
Any

The created object.
Source code in src/hyperadmin/core/adapters.py 
153
154
155
156
157
158
159
160
161
162
163
164
@abstractmethod
async def create(self, data: dict[str, Any]) -> Any:
    """
    Creates a new object.

    Args:
        data: A dictionary of data for the new object.

    Returns:
        The created object.
    """
    raise NotImplementedError

delete(pk) abstractmethod async

Deletes an object by its primary key.

Parameters:

Name Type Description Default
pk Any

The primary key of the object to delete.

 required
Source code in src/hyperadmin/core/adapters.py 
180
181
182
183
184
185
186
187
188
@abstractmethod
async def delete(self, pk: Any) -> None:
    """
    Deletes an object by its primary key.

    Args:
        pk: The primary key of the object to delete.
    """
    raise NotImplementedError

get(pk) abstractmethod async

Retrieves a single object by its primary key.

Parameters:

Name Type Description Default
pk Any

The primary key of the object to retrieve.

 required

Returns:

Type Description
Any

The retrieved object, or None if not found.
Source code in src/hyperadmin/core/adapters.py 
115
116
117
118
119
120
121
122
123
124
125
126
@abstractmethod
async def get(self, pk: Any) -> Any:
    """
    Retrieves a single object by its primary key.

    Args:
        pk: The primary key of the object to retrieve.

    Returns:
        The retrieved object, or None if not found.
    """
    raise NotImplementedError

get_choices(field, q='', limit=50, offset=0, **filters) abstractmethod async

Return paginated, searchable choices for a relation field.

Parameters:

Name Type Description Default
field str

The relation field name on this adapter's model.

 required
q str

Optional search string (ILIKE match on string columns).

 ''
limit int

Max results to return. Must not exceed 200.

 50
offset int

Number of rows to skip for pagination.

 0
**filters Any

Extra equality filters forwarded to the query (cascading support).

 {}

Returns:

Type Description
list[ChoiceItem]

A list of ChoiceItem dicts with value, label, selected=False.

Raises:

Type Description
ValueError

When limit exceeds 200.
Source code in src/hyperadmin/core/adapters.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
@abstractmethod
async def get_choices(
    self,
    field: str,
    q: str = "",
    limit: int = 50,
    offset: int = 0,
    **filters: Any,
) -> builtins.list[ChoiceItem]:
    """Return paginated, searchable choices for a relation field.

    Args:
        field: The relation field name on this adapter's model.
        q: Optional search string (ILIKE match on string columns).
        limit: Max results to return. Must not exceed 200.
        offset: Number of rows to skip for pagination.
        **filters: Extra equality filters forwarded to the query (cascading support).

    Returns:
        A list of ``ChoiceItem`` dicts with ``value``, ``label``, ``selected=False``.

    Raises:
        ValueError: When ``limit`` exceeds 200.
    """
    raise NotImplementedError

get_queryset(request=None)

Return additional equality filters merged into list() and get() queries.

The returned mapping has the shape {column_name: value} and is applied as equality predicates before any view-layer filters. Use this hook to implement features like row-level security or tenant scoping without leaking those concerns into the view layer.

Subclasses may override this method directly, or callers may register a per-request callable via :meth:set_queryset_filter. The callable is re-evaluated on every list() / get() — no caching — so it must be pure or request-scoped.

Default behaviour returns an empty dict, which is a no-op (backward compatible).

Parameters:

Name Type Description Default
request Request | None

The active Starlette/FastAPI request, when available. None is permitted so the hook is callable from contexts without an HTTP request (e.g. management scripts, tests).

 None

Returns:

Type Description
dict[str, Any]

A dict of column-name to value pairs to be ANDed into the query's WHERE
dict[str, Any]

clause. Returns {} by default.
Source code in src/hyperadmin/core/adapters.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def get_queryset(self, request: Request | None = None) -> dict[str, Any]:
    """Return additional equality filters merged into ``list()`` and ``get()`` queries.

    The returned mapping has the shape ``{column_name: value}`` and is applied as
    equality predicates **before** any view-layer filters. Use this hook to implement
    features like row-level security or tenant scoping without leaking those concerns
    into the view layer.

    Subclasses may override this method directly, or callers may register a
    per-request callable via :meth:`set_queryset_filter`. The callable is
    re-evaluated on every ``list()`` / ``get()`` — no caching — so it must be pure
    or request-scoped.

    Default behaviour returns an empty dict, which is a no-op (backward compatible).

    Args:
        request: The active Starlette/FastAPI request, when available. ``None`` is
            permitted so the hook is callable from contexts without an HTTP request
            (e.g. management scripts, tests).

    Returns:
        A dict of column-name to value pairs to be ANDed into the query's WHERE
        clause. Returns ``{}`` by default.
    """
    if self._queryset_filter is None:
        return {}
    return self._queryset_filter(request)

Retrieves related objects for a given object and field.

Parameters:

Name Type Description Default
pk Any

The primary key of the object.

 required
field str

The name of the relationship field.

 required

Returns:

Type Description
list[Any]

A list of related objects.
Source code in src/hyperadmin/core/adapters.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
@abstractmethod
async def get_related(self, pk: Any, field: str) -> builtins.list[Any]:
    """
    Retrieves related objects for a given object and field.

    Args:
        pk: The primary key of the object.
        field: The name of the relationship field.

    Returns:
        A list of related objects.
    """
    raise NotImplementedError

get_schema() abstractmethod async

Retrieves the schema definition for the model.

Returns:

Type Description
dict[str, Any]

A dictionary representing the model's schema.
Source code in src/hyperadmin/core/adapters.py 
204
205
206
207
208
209
210
211
212
@abstractmethod
async def get_schema(self) -> dict[str, Any]:
    """
    Retrieves the schema definition for the model.

    Returns:
        A dictionary representing the model's schema.
    """
    raise NotImplementedError

list(page=1, page_size=10, search=None, filters=None, order_by=None, search_fields=None) abstractmethod async

Retrieves a list of objects with optional pagination, searching, and filtering.

Parameters:

Name Type Description Default
page int

The page number for pagination.

 1
page_size int

The number of items per page.

 10
search str | None

A search query to filter the results.

 None
filters dict[str, Any] | None

A dictionary of filters to apply to the query.

 None
order_by str | None

The field to order the results by.

 None

Returns:

Type Description
tuple[list[Any], int]

A tuple containing the list of objects and the total count of objects.
Source code in src/hyperadmin/core/adapters.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@abstractmethod
async def list(
    self,
    page: int = 1,
    page_size: int = 10,
    search: str | None = None,
    filters: dict[str, Any] | None = None,
    order_by: str | None = None,
    search_fields: builtins.list[str] | None = None,
) -> tuple[builtins.list[Any], int]:
    """
    Retrieves a list of objects with optional pagination, searching, and filtering.

    Args:
        page: The page number for pagination.
        page_size: The number of items per page.
        search: A search query to filter the results.
        filters: A dictionary of filters to apply to the query.
        order_by: The field to order the results by.

    Returns:
        A tuple containing the list of objects and the total count of objects.
    """
    raise NotImplementedError

save_inline_rows(spec, rows, parent_pk) abstractmethod async

Persist validated inline rows — create, update, or delete as needed.

Parameters:

Name Type Description Default
spec InlineModelSpec

The InlineModelSpec describing the related model and FK field.

 required
rows list[dict[str, Any]]

Validated row dicts, each optionally containing _pk (for update/delete) and _delete (for deletion).

 required
parent_pk Any

The primary key of the parent object to associate new rows with.

 required
Source code in src/hyperadmin/core/adapters.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
@abstractmethod
async def save_inline_rows(
    self,
    spec: InlineModelSpec,
    rows: builtins.list[dict[str, Any]],
    parent_pk: Any,
) -> None:
    """Persist validated inline rows — create, update, or delete as needed.

    Args:
        spec: The ``InlineModelSpec`` describing the related model and FK field.
        rows: Validated row dicts, each optionally containing ``_pk`` (for
            update/delete) and ``_delete`` (for deletion).
        parent_pk: The primary key of the parent object to associate new rows with.
    """
    raise NotImplementedError

set_queryset_filter(filter_fn)

Register a per-request callable that returns additional queryset filters.

The callable receives the current :class:starlette.requests.Request (or None) and must return a dict of {column_name: value} equality filters. It is re-invoked on every list() / get() so it may consult request-scoped state (e.g. request.state.user).

Parameters:

Name Type Description Default
filter_fn QuerysetFilter

A callable taking an optional :class:Request and returning a dict of equality filters.

 required
Source code in src/hyperadmin/core/adapters.py 
87
88
89
90
91
92
93
94
95
96
97
98
99
def set_queryset_filter(self, filter_fn: QuerysetFilter) -> None:
    """Register a per-request callable that returns additional queryset filters.

    The callable receives the current :class:`starlette.requests.Request` (or
    ``None``) and must return a dict of ``{column_name: value}`` equality filters.
    It is re-invoked on every ``list()`` / ``get()`` so it may consult
    request-scoped state (e.g. ``request.state.user``).

    Args:
        filter_fn: A callable taking an optional :class:`Request` and returning a
            dict of equality filters.
    """
    self._queryset_filter = filter_fn

update(pk, data) abstractmethod async

Updates an existing object.

Parameters:

Name Type Description Default
pk Any

The primary key of the object to update.

 required
data dict[str, Any]

A dictionary of data to update the object with.

 required

Returns:

Type Description
Any

The updated object.
Source code in src/hyperadmin/core/adapters.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
@abstractmethod
async def update(self, pk: Any, data: dict[str, Any]) -> Any:
    """
    Updates an existing object.

    Args:
        pk: The primary key of the object to update.
        data: A dictionary of data to update the object with.

    Returns:
        The updated object.
    """
    raise NotImplementedError

hyperadmin.adapters.sqlmodel.SQLModelAdapter

Bases: BaseAdapter

Data adapter for SQLModel.

Source code in src/hyperadmin/adapters/sqlmodel.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
class SQLModelAdapter(BaseAdapter):
    """
    Data adapter for SQLModel.
    """

    def __init__(self, model: type[SQLModel], engine: AsyncEngine):
        super().__init__(model, engine)
        self.inspector = inspect(model)

    async def get(self, pk: Any) -> Any:
        """
        Retrieves a single object by its primary key.

        Any filters returned by :meth:`get_queryset` are merged into the WHERE clause
        before the primary-key predicate so excluded rows resolve to ``None``.

        Args:
            pk: The primary key of the object to retrieve.

        Returns:
            The retrieved object, or None if not found.
        """
        queryset_filters = self._resolve_queryset_filters()
        async with AsyncSession(self.engine) as session:
            mapper = inspect(self.model)
            options = [selectinload(getattr(self.model, rel.key)) for rel in mapper.relationships]
            query = select(self.model).where(self.model.id == pk).options(*options)
            for key, value in queryset_filters.items():
                query = query.where(getattr(self.model, key) == value)
            result = await session.execute(query)
            return result.scalar_one_or_none()

    async def list(
        self,
        page: int = 1,
        page_size: int = 10,
        search: str | None = None,
        filters: dict[str, Any] | None = None,
        order_by: str | None = None,
        search_fields: list[str] | None = None,
    ) -> tuple[list[Any], int]:
        """
        Retrieves a list of objects with optional pagination, searching, and filtering.

        Filters returned by :meth:`get_queryset` are applied **before** any
        view-layer ``filters`` so they cannot be bypassed. The same predicates are
        included in the count query, keeping pagination math consistent.
        """
        queryset_filters = self._resolve_queryset_filters()
        async with AsyncSession(self.engine) as session:
            query = select(self.model)

            # Eagerly load all relationships to avoid DetachedInstanceError
            mapper = inspect(self.model)
            for rel in mapper.relationships:
                query = query.options(selectinload(getattr(self.model, rel.key)))

            # Apply queryset hook filters first (RLS / tenant scoping)
            for key, value in queryset_filters.items():
                query = query.where(getattr(self.model, key) == value)

            # Apply filtering
            if filters:
                for key, value in filters.items():
                    query = query.where(getattr(self.model, key) == value)

            # Apply searching using configured search_fields
            if search:
                fields_to_search = search_fields or self._detect_search_fields()
                if fields_to_search:
                    conditions = []
                    for field_name in fields_to_search:
                        col = getattr(self.model, field_name, None)
                        if col is not None:
                            conditions.append(col.ilike(f"%{search}%"))
                    if conditions:
                        query = query.where(or_(*conditions))

            # Apply ordering
            if order_by:
                if order_by.startswith("-"):
                    query = query.order_by(getattr(self.model, order_by[1:]).desc())
                else:
                    query = query.order_by(getattr(self.model, order_by).asc())

            # Get total count
            count_query = select(func.count()).select_from(query.subquery())
            total_count_result = await session.execute(count_query)
            total_count = total_count_result.scalar_one()

            # Apply pagination
            offset = (page - 1) * page_size
            query = query.offset(offset).limit(page_size)

            # Get the rows
            results = await session.execute(query)
            return list(results.scalars().all()), total_count

    def _detect_search_fields(self) -> builtins.list[str]:
        """Detect string columns on the model for search fallback."""
        mapper: Any = self.inspector
        return [
            col.key
            for col in mapper.columns
            if isinstance(col.type, (String, AutoString)) and not col.primary_key
        ]

    async def create(self, data: dict[str, Any]) -> Any:
        """
        Creates a new object.

        Args:
            data: A dictionary of data for the new object.

        Returns:
            The created object.
        """
        async with AsyncSession(self.engine) as session:
            db_obj = self.model(**data)
            session.add(db_obj)
            await session.commit()
            await session.refresh(db_obj)
            return db_obj

    async def update(self, pk: Any, data: dict[str, Any]) -> Any:
        """
        Updates an existing object.

        Args:
            pk: The primary key of the object to update.
            data: A dictionary of data to update the object with.

        Returns:
            The updated object.
        """
        async with AsyncSession(self.engine) as session:
            db_obj = await session.get(self.model, pk)
            if db_obj:
                for key, value in data.items():
                    setattr(db_obj, key, value)
                session.add(db_obj)
                await session.commit()
                await session.refresh(db_obj)
            return db_obj

    async def delete(self, pk: Any) -> None:
        """
        Deletes an object.

        Args:
            pk: The primary key of the object to delete.
        """
        async with AsyncSession(self.engine) as session:
            db_obj = await session.get(self.model, pk)
            if db_obj:
                await session.delete(db_obj)
                await session.commit()

    async def get_related(self, pk: Any, field: str) -> builtins.list[Any]:
        """
        Retrieves related objects for a given object and field.

        Args:
            pk: The primary key of the object.
            field: The name of the related field.

        Returns:
            A list of related objects.
        """
        async with AsyncSession(self.engine) as session:
            query = (
                select(self.model)
                .where(self.model.id == pk)
                .options(selectinload(getattr(self.model, field)))
            )
            result = await session.execute(query)
            db_obj = result.scalar_one_or_none()
            if db_obj:
                return getattr(db_obj, field)
            return []

    async def get_schema(self) -> dict[str, Any]:
        """
        Returns the JSON schema for the model.

        Returns:
            A dictionary representing the JSON schema.
        """
        return self.model.model_json_schema()

    async def get_choices(
        self,
        field: str,
        q: str = "",
        limit: int = 50,
        offset: int = 0,
        **filters: Any,
    ) -> builtins.list[ChoiceItem]:
        """Return paginated, searchable choices for a relation field.

        Performs a single SELECT on the related model — no N+1.
        """
        if limit > _MAX_CHOICES_LIMIT:
            raise ValueError(f"limit {limit} exceeds maximum of {_MAX_CHOICES_LIMIT}")

        mapper = inspect(self.model)
        target_model = None
        for rel in mapper.relationships:
            if rel.key == field:
                target_model = rel.mapper.class_
                break

        if target_model is None:
            return []

        target_inspector = inspect(target_model)
        async with AsyncSession(self.engine) as session:
            query = select(target_model)

            if q:
                str_cols = [
                    c for c in target_inspector.c if isinstance(c.type, AutoString | String)
                ]
                if str_cols:
                    query = query.where(or_(*[c.ilike(f"%{q}%") for c in str_cols[:3]]))

            for key, value in filters.items():
                if hasattr(target_model, key):
                    query = query.where(getattr(target_model, key) == value)

            query = query.offset(offset).limit(limit)
            result = await session.execute(query)
            items = result.scalars().all()

        return [
            ChoiceItem(
                value=str(getattr(item, "id", "")),
                label=str(item),
                selected=False,
            )
            for item in items
        ]

    async def save_inline_rows(
        self,
        spec: InlineModelSpec,
        rows: builtins.list[dict[str, Any]],
        parent_pk: Any,
    ) -> None:
        """Persist validated inline rows — create, update, or delete as needed.

        A fresh ``SQLModelAdapter`` is constructed for the inline model using
        this adapter's engine so all operations share the same database connection.

        Args:
            spec: The ``InlineModelSpec`` describing the related model and FK field.
            rows: Validated row dicts, each optionally containing ``_pk`` (for
                update/delete) and ``_delete`` (for deletion).
            parent_pk: The primary key of the parent object to associate new rows with.
        """
        inline_adapter = SQLModelAdapter(spec.model, self.engine)
        for row in rows:
            if row.get("_delete") and row.get("_pk"):
                await inline_adapter.delete(pk=row["_pk"])
            elif "_pk" in row:
                pk = row["_pk"]
                row_data = {k: v for k, v in row.items() if k not in ("_pk", "_delete")}
                await inline_adapter.update(pk=pk, data=row_data)
            else:
                row_data = {k: v for k, v in row.items() if k not in ("_pk", "_delete")}
                row_data[spec.fk_field] = parent_pk
                await inline_adapter.create(data=row_data)

create(data) async

Creates a new object.

Parameters:

Name Type Description Default
data dict[str, Any]

A dictionary of data for the new object.

 required

Returns:

Type Description
Any

The created object.
Source code in src/hyperadmin/adapters/sqlmodel.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
async def create(self, data: dict[str, Any]) -> Any:
    """
    Creates a new object.

    Args:
        data: A dictionary of data for the new object.

    Returns:
        The created object.
    """
    async with AsyncSession(self.engine) as session:
        db_obj = self.model(**data)
        session.add(db_obj)
        await session.commit()
        await session.refresh(db_obj)
        return db_obj

delete(pk) async

Deletes an object.

Parameters:

Name Type Description Default
pk Any

The primary key of the object to delete.

 required
Source code in src/hyperadmin/adapters/sqlmodel.py 
162
163
164
165
166
167
168
169
170
171
172
173
async def delete(self, pk: Any) -> None:
    """
    Deletes an object.

    Args:
        pk: The primary key of the object to delete.
    """
    async with AsyncSession(self.engine) as session:
        db_obj = await session.get(self.model, pk)
        if db_obj:
            await session.delete(db_obj)
            await session.commit()

get(pk) async

Retrieves a single object by its primary key.

Any filters returned by :meth:get_queryset are merged into the WHERE clause before the primary-key predicate so excluded rows resolve to None.

Parameters:

Name Type Description Default
pk Any

The primary key of the object to retrieve.

 required

Returns:

Type Description
Any

The retrieved object, or None if not found.
Source code in src/hyperadmin/adapters/sqlmodel.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
async def get(self, pk: Any) -> Any:
    """
    Retrieves a single object by its primary key.

    Any filters returned by :meth:`get_queryset` are merged into the WHERE clause
    before the primary-key predicate so excluded rows resolve to ``None``.

    Args:
        pk: The primary key of the object to retrieve.

    Returns:
        The retrieved object, or None if not found.
    """
    queryset_filters = self._resolve_queryset_filters()
    async with AsyncSession(self.engine) as session:
        mapper = inspect(self.model)
        options = [selectinload(getattr(self.model, rel.key)) for rel in mapper.relationships]
        query = select(self.model).where(self.model.id == pk).options(*options)
        for key, value in queryset_filters.items():
            query = query.where(getattr(self.model, key) == value)
        result = await session.execute(query)
        return result.scalar_one_or_none()

get_choices(field, q='', limit=50, offset=0, **filters) async

Return paginated, searchable choices for a relation field.

Performs a single SELECT on the related model — no N+1.

Source code in src/hyperadmin/adapters/sqlmodel.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
async def get_choices(
    self,
    field: str,
    q: str = "",
    limit: int = 50,
    offset: int = 0,
    **filters: Any,
) -> builtins.list[ChoiceItem]:
    """Return paginated, searchable choices for a relation field.

    Performs a single SELECT on the related model — no N+1.
    """
    if limit > _MAX_CHOICES_LIMIT:
        raise ValueError(f"limit {limit} exceeds maximum of {_MAX_CHOICES_LIMIT}")

    mapper = inspect(self.model)
    target_model = None
    for rel in mapper.relationships:
        if rel.key == field:
            target_model = rel.mapper.class_
            break

    if target_model is None:
        return []

    target_inspector = inspect(target_model)
    async with AsyncSession(self.engine) as session:
        query = select(target_model)

        if q:
            str_cols = [
                c for c in target_inspector.c if isinstance(c.type, AutoString | String)
            ]
            if str_cols:
                query = query.where(or_(*[c.ilike(f"%{q}%") for c in str_cols[:3]]))

        for key, value in filters.items():
            if hasattr(target_model, key):
                query = query.where(getattr(target_model, key) == value)

        query = query.offset(offset).limit(limit)
        result = await session.execute(query)
        items = result.scalars().all()

    return [
        ChoiceItem(
            value=str(getattr(item, "id", "")),
            label=str(item),
            selected=False,
        )
        for item in items
    ]

Retrieves related objects for a given object and field.

Parameters:

Name Type Description Default
pk Any

The primary key of the object.

 required
field str

The name of the related field.

 required

Returns:

Type Description
list[Any]

A list of related objects.
Source code in src/hyperadmin/adapters/sqlmodel.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
async def get_related(self, pk: Any, field: str) -> builtins.list[Any]:
    """
    Retrieves related objects for a given object and field.

    Args:
        pk: The primary key of the object.
        field: The name of the related field.

    Returns:
        A list of related objects.
    """
    async with AsyncSession(self.engine) as session:
        query = (
            select(self.model)
            .where(self.model.id == pk)
            .options(selectinload(getattr(self.model, field)))
        )
        result = await session.execute(query)
        db_obj = result.scalar_one_or_none()
        if db_obj:
            return getattr(db_obj, field)
        return []

get_schema() async

Returns the JSON schema for the model.

Returns:

Type Description
dict[str, Any]

A dictionary representing the JSON schema.
Source code in src/hyperadmin/adapters/sqlmodel.py 
198
199
200
201
202
203
204
205
async def get_schema(self) -> dict[str, Any]:
    """
    Returns the JSON schema for the model.

    Returns:
        A dictionary representing the JSON schema.
    """
    return self.model.model_json_schema()

list(page=1, page_size=10, search=None, filters=None, order_by=None, search_fields=None) async

Retrieves a list of objects with optional pagination, searching, and filtering.

Filters returned by :meth:get_queryset are applied before any view-layer filters so they cannot be bypassed. The same predicates are included in the count query, keeping pagination math consistent.

Source code in src/hyperadmin/adapters/sqlmodel.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
async def list(
    self,
    page: int = 1,
    page_size: int = 10,
    search: str | None = None,
    filters: dict[str, Any] | None = None,
    order_by: str | None = None,
    search_fields: list[str] | None = None,
) -> tuple[list[Any], int]:
    """
    Retrieves a list of objects with optional pagination, searching, and filtering.

    Filters returned by :meth:`get_queryset` are applied **before** any
    view-layer ``filters`` so they cannot be bypassed. The same predicates are
    included in the count query, keeping pagination math consistent.
    """
    queryset_filters = self._resolve_queryset_filters()
    async with AsyncSession(self.engine) as session:
        query = select(self.model)

        # Eagerly load all relationships to avoid DetachedInstanceError
        mapper = inspect(self.model)
        for rel in mapper.relationships:
            query = query.options(selectinload(getattr(self.model, rel.key)))

        # Apply queryset hook filters first (RLS / tenant scoping)
        for key, value in queryset_filters.items():
            query = query.where(getattr(self.model, key) == value)

        # Apply filtering
        if filters:
            for key, value in filters.items():
                query = query.where(getattr(self.model, key) == value)

        # Apply searching using configured search_fields
        if search:
            fields_to_search = search_fields or self._detect_search_fields()
            if fields_to_search:
                conditions = []
                for field_name in fields_to_search:
                    col = getattr(self.model, field_name, None)
                    if col is not None:
                        conditions.append(col.ilike(f"%{search}%"))
                if conditions:
                    query = query.where(or_(*conditions))

        # Apply ordering
        if order_by:
            if order_by.startswith("-"):
                query = query.order_by(getattr(self.model, order_by[1:]).desc())
            else:
                query = query.order_by(getattr(self.model, order_by).asc())

        # Get total count
        count_query = select(func.count()).select_from(query.subquery())
        total_count_result = await session.execute(count_query)
        total_count = total_count_result.scalar_one()

        # Apply pagination
        offset = (page - 1) * page_size
        query = query.offset(offset).limit(page_size)

        # Get the rows
        results = await session.execute(query)
        return list(results.scalars().all()), total_count

save_inline_rows(spec, rows, parent_pk) async

Persist validated inline rows — create, update, or delete as needed.

A fresh SQLModelAdapter is constructed for the inline model using this adapter's engine so all operations share the same database connection.

Parameters:

Name Type Description Default
spec InlineModelSpec

The InlineModelSpec describing the related model and FK field.

 required
rows list[dict[str, Any]]

Validated row dicts, each optionally containing _pk (for update/delete) and _delete (for deletion).

 required
parent_pk Any

The primary key of the parent object to associate new rows with.

 required
Source code in src/hyperadmin/adapters/sqlmodel.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
async def save_inline_rows(
    self,
    spec: InlineModelSpec,
    rows: builtins.list[dict[str, Any]],
    parent_pk: Any,
) -> None:
    """Persist validated inline rows — create, update, or delete as needed.

    A fresh ``SQLModelAdapter`` is constructed for the inline model using
    this adapter's engine so all operations share the same database connection.

    Args:
        spec: The ``InlineModelSpec`` describing the related model and FK field.
        rows: Validated row dicts, each optionally containing ``_pk`` (for
            update/delete) and ``_delete`` (for deletion).
        parent_pk: The primary key of the parent object to associate new rows with.
    """
    inline_adapter = SQLModelAdapter(spec.model, self.engine)
    for row in rows:
        if row.get("_delete") and row.get("_pk"):
            await inline_adapter.delete(pk=row["_pk"])
        elif "_pk" in row:
            pk = row["_pk"]
            row_data = {k: v for k, v in row.items() if k not in ("_pk", "_delete")}
            await inline_adapter.update(pk=pk, data=row_data)
        else:
            row_data = {k: v for k, v in row.items() if k not in ("_pk", "_delete")}
            row_data[spec.fk_field] = parent_pk
            await inline_adapter.create(data=row_data)

update(pk, data) async

Updates an existing object.

Parameters:

Name Type Description Default
pk Any

The primary key of the object to update.

 required
data dict[str, Any]

A dictionary of data to update the object with.

 required

Returns:

Type Description
Any

The updated object.
Source code in src/hyperadmin/adapters/sqlmodel.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
async def update(self, pk: Any, data: dict[str, Any]) -> Any:
    """
    Updates an existing object.

    Args:
        pk: The primary key of the object to update.
        data: A dictionary of data to update the object with.

    Returns:
        The updated object.
    """
    async with AsyncSession(self.engine) as session:
        db_obj = await session.get(self.model, pk)
        if db_obj:
            for key, value in data.items():
                setattr(db_obj, key, value)
            session.add(db_obj)
            await session.commit()
            await session.refresh(db_obj)
        return db_obj

hyperadmin.adapters.sqlalchemy.SQLAlchemyAdapter

Bases: BaseAdapter

Source code in src/hyperadmin/adapters/sqlalchemy.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
class SQLAlchemyAdapter(BaseAdapter):
    def __init__(self, model: type[SQLModel], engine: AsyncEngine) -> None:
        super().__init__(model=model, engine=engine)
        self.inspector = inspect(model)
        if self.inspector is None:
            raise ValueError("Could not inspect model. Is it a valid SQLAlchemy model?")

    async def get(self, pk: Any) -> Any:
        """Retrieve an object by primary key, applying any ``get_queryset`` filters.

        Filters returned by :meth:`get_queryset` are ANDed with the primary-key
        predicate so excluded rows resolve to ``None`` (not raised).
        """
        queryset_filters = self._resolve_queryset_filters()
        async with AsyncSession(self.engine) as session:
            if not queryset_filters:
                return await session.get(self.model, pk)
            assert self.inspector is not None  # noqa: S101 — validated in __init__
            pk_col = self.inspector.primary_key[0]
            conditions = [pk_col == pk]
            for key, value in queryset_filters.items():
                conditions.append(getattr(self.model, key) == value)
            query = select(self.model).where(and_(*conditions))
            result = await session.execute(query)
            return result.scalar_one_or_none()

    async def list(
        self,
        page: int = 1,
        page_size: int = 10,
        search: str | None = None,
        filters: dict[str, Any] | None = None,
        order_by: str | None = None,
        search_fields: list[str] | None = None,  # noqa: ARG002
    ) -> tuple[list[Any], int]:
        queryset_filters = self._resolve_queryset_filters()
        where_conditions = [
            getattr(self.model, key) == value for key, value in queryset_filters.items()
        ]
        if filters:
            for key, value in filters.items():
                where_conditions.append(getattr(self.model, key) == value)

        if search and self.inspector:
            search_clauses = [
                getattr(self.model, column.name).ilike(f"%{search}%")
                for column in self.inspector.c
                if isinstance(column.type, AutoString | String)
            ]
            if search_clauses:
                where_conditions.append(or_(*search_clauses))

        items_query = select(self.model)
        if where_conditions:
            items_query = items_query.where(and_(*where_conditions))

        count_query = select(func.count()).select_from(self.model)
        if where_conditions:
            count_query = count_query.where(and_(*where_conditions))

        if order_by:
            if order_by.startswith("-"):
                items_query = items_query.order_by(getattr(self.model, order_by[1:]).desc())
            else:
                items_query = items_query.order_by(getattr(self.model, order_by).asc())

        paginated_items_query = items_query.offset((page - 1) * page_size).limit(page_size)

        async with AsyncSession(self.engine) as session:
            total_count_result = await session.execute(count_query)
            total_count = total_count_result.scalar_one()

            items_result = await session.execute(paginated_items_query)
            items = items_result.scalars().all()

            return list(items), total_count

    async def create(self, data: dict[str, Any]) -> Any:
        db_obj = self.model.model_validate(data)
        async with AsyncSession(self.engine) as session:
            session.add(db_obj)
            await session.commit()
            await session.refresh(db_obj)
        return db_obj

    async def update(self, pk: Any, data: dict[str, Any]) -> Any:
        async with AsyncSession(self.engine) as session:
            db_obj = await session.get(self.model, pk)
            if db_obj:
                for key, value in data.items():
                    setattr(db_obj, key, value)
                session.add(db_obj)
                await session.commit()
                await session.refresh(db_obj)
            return db_obj

    async def delete(self, pk: Any) -> None:
        async with AsyncSession(self.engine) as session:
            db_obj = await session.get(self.model, pk)
            if db_obj:
                await session.delete(db_obj)
                await session.commit()

    async def get_related(self, pk: Any, field: str) -> builtins.list[Any]:
        if not self.inspector or not hasattr(self.model, field):
            return []
        async with AsyncSession(self.engine) as session:
            query = (
                select(self.model)
                .where(self.inspector.primary_key[0] == pk)
                .options(selectinload(getattr(self.model, field)))
            )
            result = await session.execute(query)
            db_obj = result.scalar_one_or_none()

            if not db_obj:
                return []

            try:
                related = getattr(db_obj, field)
                if related is None:
                    return []
                if isinstance(related, list):
                    return list(related)
                return [related]
            except AttributeError:
                return []

    async def get_schema(self) -> dict[str, Any]:
        return self.model.model_json_schema()

    async def get_choices(
        self,
        field: str,
        q: str = "",
        limit: int = 50,
        offset: int = 0,
        **filters: Any,
    ) -> builtins.list[ChoiceItem]:
        """Return paginated, searchable choices for a relation field.

        Performs a single SELECT on the related model — no N+1.
        """
        if limit > _MAX_CHOICES_LIMIT:
            raise ValueError(f"limit {limit} exceeds maximum of {_MAX_CHOICES_LIMIT}")

        if not self.inspector:
            return []

        target_model = None
        for rel in self.inspector.relationships:
            if rel.key == field:
                target_model = rel.mapper.class_
                break

        if target_model is None:
            return []

        target_inspector = inspect(target_model)
        async with AsyncSession(self.engine) as session:
            query = select(target_model)

            if q:
                str_cols = [
                    c for c in target_inspector.c if isinstance(c.type, AutoString | String)
                ]
                if str_cols:
                    query = query.where(or_(*[c.ilike(f"%{q}%") for c in str_cols[:3]]))

            for key, value in filters.items():
                if hasattr(target_model, key):
                    query = query.where(getattr(target_model, key) == value)

            query = query.offset(offset).limit(limit)
            result = await session.execute(query)
            items = result.scalars().all()

        return [
            ChoiceItem(
                value=str(getattr(item, "id", "")),
                label=str(item),
                selected=False,
            )
            for item in items
        ]

    async def save_inline_rows(
        self,
        spec: InlineModelSpec,
        rows: builtins.list[dict[str, Any]],
        parent_pk: Any,
    ) -> None:
        """Persist validated inline rows — create, update, or delete as needed.

        Args:
            spec: The ``InlineModelSpec`` describing the related model and FK field.
            rows: Validated row dicts, each optionally containing ``_pk`` (for
                update/delete) and ``_delete`` (for deletion).
            parent_pk: The primary key of the parent object to associate new rows with.
        """
        inline_adapter = SQLAlchemyAdapter(spec.model, self.engine)
        for row in rows:
            if row.get("_delete") and row.get("_pk"):
                await inline_adapter.delete(pk=row["_pk"])
            elif "_pk" in row:
                pk = row["_pk"]
                row_data = {k: v for k, v in row.items() if k not in ("_pk", "_delete")}
                await inline_adapter.update(pk=pk, data=row_data)
            else:
                row_data = {k: v for k, v in row.items() if k not in ("_pk", "_delete")}
                row_data[spec.fk_field] = parent_pk
                await inline_adapter.create(data=row_data)

get(pk) async

Retrieve an object by primary key, applying any get_queryset filters.

Filters returned by :meth:get_queryset are ANDed with the primary-key predicate so excluded rows resolve to None (not raised).

Source code in src/hyperadmin/adapters/sqlalchemy.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
async def get(self, pk: Any) -> Any:
    """Retrieve an object by primary key, applying any ``get_queryset`` filters.

    Filters returned by :meth:`get_queryset` are ANDed with the primary-key
    predicate so excluded rows resolve to ``None`` (not raised).
    """
    queryset_filters = self._resolve_queryset_filters()
    async with AsyncSession(self.engine) as session:
        if not queryset_filters:
            return await session.get(self.model, pk)
        assert self.inspector is not None  # noqa: S101 — validated in __init__
        pk_col = self.inspector.primary_key[0]
        conditions = [pk_col == pk]
        for key, value in queryset_filters.items():
            conditions.append(getattr(self.model, key) == value)
        query = select(self.model).where(and_(*conditions))
        result = await session.execute(query)
        return result.scalar_one_or_none()

get_choices(field, q='', limit=50, offset=0, **filters) async

Return paginated, searchable choices for a relation field.

Performs a single SELECT on the related model — no N+1.

Source code in src/hyperadmin/adapters/sqlalchemy.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
async def get_choices(
    self,
    field: str,
    q: str = "",
    limit: int = 50,
    offset: int = 0,
    **filters: Any,
) -> builtins.list[ChoiceItem]:
    """Return paginated, searchable choices for a relation field.

    Performs a single SELECT on the related model — no N+1.
    """
    if limit > _MAX_CHOICES_LIMIT:
        raise ValueError(f"limit {limit} exceeds maximum of {_MAX_CHOICES_LIMIT}")

    if not self.inspector:
        return []

    target_model = None
    for rel in self.inspector.relationships:
        if rel.key == field:
            target_model = rel.mapper.class_
            break

    if target_model is None:
        return []

    target_inspector = inspect(target_model)
    async with AsyncSession(self.engine) as session:
        query = select(target_model)

        if q:
            str_cols = [
                c for c in target_inspector.c if isinstance(c.type, AutoString | String)
            ]
            if str_cols:
                query = query.where(or_(*[c.ilike(f"%{q}%") for c in str_cols[:3]]))

        for key, value in filters.items():
            if hasattr(target_model, key):
                query = query.where(getattr(target_model, key) == value)

        query = query.offset(offset).limit(limit)
        result = await session.execute(query)
        items = result.scalars().all()

    return [
        ChoiceItem(
            value=str(getattr(item, "id", "")),
            label=str(item),
            selected=False,
        )
        for item in items
    ]

save_inline_rows(spec, rows, parent_pk) async

Persist validated inline rows — create, update, or delete as needed.

Parameters:

Name Type Description Default
spec InlineModelSpec

The InlineModelSpec describing the related model and FK field.

 required
rows list[dict[str, Any]]

Validated row dicts, each optionally containing _pk (for update/delete) and _delete (for deletion).

 required
parent_pk Any

The primary key of the parent object to associate new rows with.

 required
Source code in src/hyperadmin/adapters/sqlalchemy.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
async def save_inline_rows(
    self,
    spec: InlineModelSpec,
    rows: builtins.list[dict[str, Any]],
    parent_pk: Any,
) -> None:
    """Persist validated inline rows — create, update, or delete as needed.

    Args:
        spec: The ``InlineModelSpec`` describing the related model and FK field.
        rows: Validated row dicts, each optionally containing ``_pk`` (for
            update/delete) and ``_delete`` (for deletion).
        parent_pk: The primary key of the parent object to associate new rows with.
    """
    inline_adapter = SQLAlchemyAdapter(spec.model, self.engine)
    for row in rows:
        if row.get("_delete") and row.get("_pk"):
            await inline_adapter.delete(pk=row["_pk"])
        elif "_pk" in row:
            pk = row["_pk"]
            row_data = {k: v for k, v in row.items() if k not in ("_pk", "_delete")}
            await inline_adapter.update(pk=pk, data=row_data)
        else:
            row_data = {k: v for k, v in row.items() if k not in ("_pk", "_delete")}
            row_data[spec.fk_field] = parent_pk
            await inline_adapter.create(data=row_data)