From 349798dd903d31183c03c2541989b195eec67be4 Mon Sep 17 00:00:00 2001 From: Maxim Dementyev Date: Tue, 7 Apr 2026 21:00:20 +0300 Subject: [PATCH 1/4] Test implementation multiple admin panels --- sqladmin/_menu.py | 6 +- sqladmin/application.py | 19 ++-- sqladmin/authentication.py | 6 +- sqladmin/helpers.py | 29 +++++ sqladmin/models.py | 23 ++-- sqladmin/templates/sqladmin/base.html | 26 ++--- sqladmin/templates/sqladmin/create.html | 4 +- sqladmin/templates/sqladmin/details.html | 8 +- sqladmin/templates/sqladmin/edit.html | 4 +- sqladmin/templates/sqladmin/layout.html | 4 +- sqladmin/templates/sqladmin/list.html | 16 +-- sqladmin/templates/sqladmin/login.html | 2 +- sqladmin/templating.py | 4 +- tests/test_ajax.py | 8 +- tests/test_application.py | 10 +- tests/test_authentication.py | 3 +- tests/test_forms/test_forms.py | 2 +- tests/test_menu.py | 10 +- tests/test_models.py | 128 ++++++++++++++--------- tests/test_models_action.py | 14 +-- tests/test_pretty_export.py | 39 +++---- 21 files changed, 214 insertions(+), 151 deletions(-) diff --git a/sqladmin/_menu.py b/sqladmin/_menu.py index 057528b5..a4d8cab6 100644 --- a/sqladmin/_menu.py +++ b/sqladmin/_menu.py @@ -8,6 +8,8 @@ if TYPE_CHECKING: from sqladmin.application import BaseView, ModelView +from sqladmin.helpers import local_url_for + class ItemMenu: def __init__(self, name: str, icon: str | None = None) -> None: @@ -73,8 +75,8 @@ def is_active(self, request: Request) -> bool: def url(self, request: Request) -> str | URL: if self.view.is_model: - return request.url_for("admin:list", identity=self.view.identity) - return request.url_for(f"admin:{self.view.identity}") + return local_url_for(request, "list", identity=self.view.identity) + return local_url_for(request, self.view.identity) @property def display_name(self) -> str: diff --git a/sqladmin/application.py b/sqladmin/application.py index ac0af92d..deb307cf 100644 --- a/sqladmin/application.py +++ b/sqladmin/application.py @@ -41,6 +41,7 @@ from sqladmin.helpers import ( get_object_identifier, is_async_session_maker, + local_url_for, slugify_action_name, ) from sqladmin.models import BaseView, ModelView @@ -236,13 +237,12 @@ class UserAdmin(ModelView, model=User): view._admin_ref = self # Set database engine from Admin instance - view.session_maker = self.session_maker view.is_async = self.is_async view.ajax_lookup_url = urljoin( self.base_url + "/", f"{view.identity}/ajax/lookup" ) view.templates = self.templates - view_instance = view() + view_instance = view(self.session_maker) self._find_decorated_funcs( view, view_instance, self._handle_action_decorated_func @@ -367,6 +367,7 @@ def __init__( # type: ignore[no-any-unimported] debug: bool = False, templates_dir: str = "templates", authentication_backend: AuthenticationBackend | None = None, + mount_name: str = "admin", ) -> None: """ Args: @@ -446,7 +447,7 @@ async def http_exception( self.admin.router.routes = routes self.admin.exception_handlers = {HTTPException: http_exception} self.admin.debug = debug - self.app.mount(base_url, app=self.admin, name="admin") + self.app.mount(base_url, app=self.admin, name=mount_name) @login_required async def index(self, request: Request) -> Response: @@ -519,7 +520,7 @@ async def delete(self, request: Request) -> Response: referer_url = URL(request.headers.get("referer", "")) referer_params = MultiDict(parse_qsl(referer_url.query)) - url = URL(str(request.url_for("admin:list", identity=identity))) + url = URL(str(local_url_for(request, "list", identity=identity))) url = url.include_query_params(**referer_params) return PlainTextResponse(content=str(url)) @@ -658,7 +659,7 @@ async def login(self, request: Request) -> Response: request, "sqladmin/login.html", context, status_code=400 ) - return RedirectResponse(request.url_for("admin:index"), status_code=302) + return RedirectResponse(local_url_for(request, "index"), status_code=302) async def logout(self, request: Request) -> Response: if self.authentication_backend is None: @@ -672,7 +673,7 @@ async def logout(self, request: Request) -> Response: if isinstance(response, Response): return response - return RedirectResponse(request.url_for("admin:index"), status_code=302) + return RedirectResponse(local_url_for(request, "index"), status_code=302) async def ajax_lookup(self, request: Request) -> Response: """Ajax lookup route.""" @@ -706,14 +707,14 @@ def get_save_redirect_url( identifier = get_object_identifier(obj) if form.get("save") == "Save": - return request.url_for("admin:list", identity=identity) + return local_url_for(request, "list", identity=identity) if form.get("save") == "Save and continue editing" or ( form.get("save") == "Save as new" and model_view.save_as_continue ): - return request.url_for("admin:edit", identity=identity, pk=identifier) + return local_url_for(request, "edit", identity=identity, pk=identifier) - return request.url_for("admin:create", identity=identity) + return local_url_for(request, "create", identity=identity) async def _handle_form_data(self, request: Request, obj: Any = None) -> FormData: """ diff --git a/sqladmin/authentication.py b/sqladmin/authentication.py index 3238402f..e4a8746b 100644 --- a/sqladmin/authentication.py +++ b/sqladmin/authentication.py @@ -8,6 +8,8 @@ from starlette.requests import Request from starlette.responses import RedirectResponse, Response +from sqladmin.helpers import local_url_for + class AuthenticationBackend: """Base class for implementing the Authentication into SQLAdmin. @@ -66,7 +68,9 @@ async def wrapper_decorator(*args: Any, **kwargs: Any) -> Any: if isinstance(response, Response): return response if not bool(response): - return RedirectResponse(request.url_for("admin:login"), status_code=302) + return RedirectResponse( + local_url_for(request, "login"), status_code=302 + ) if inspect.iscoroutinefunction(func): return await func(*args, **kwargs) diff --git a/sqladmin/helpers.py b/sqladmin/helpers.py index a7b9b320..ecc398dc 100644 --- a/sqladmin/helpers.py +++ b/sqladmin/helpers.py @@ -18,6 +18,9 @@ from sqlalchemy import Column, inspect from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import RelationshipProperty, sessionmaker +from starlette.datastructures import URL +from starlette.requests import Request +from starlette.routing import Mount, Router from sqladmin._types import MODEL_PROPERTY @@ -322,3 +325,29 @@ def choice_coerce(value: Any) -> Any: def is_async_session_maker(session_maker: sessionmaker) -> bool: return AsyncSession in session_maker.class_.__mro__ + + +def local_url_for(request: Request, name: str, **kwargs: Any) -> URL: + target_router = request.app.router + start_router = request.scope["router"] + if start_router == target_router: + return request.url_for(name, **kwargs) + router_name = get_current_router_name(start_router, target_router) + return request.url_for( + f"{f'{router_name}:' if router_name else ''}{name}", **kwargs + ) + + +def get_current_router_name(start_router: Any, target_router: Router) -> str | None: + for router in getattr(start_router, "routes", []): + if router == target_router: + return router.name + if not isinstance(router, Mount): + continue + app_router = getattr(router.app, "router", None) + if app_router == target_router: + return router.name + sub_name = get_current_router_name(router, target_router) + if sub_name is not None: + return f"{router.name}:{sub_name}" + return None diff --git a/sqladmin/models.py b/sqladmin/models.py index c1fdec49..e7c819c6 100644 --- a/sqladmin/models.py +++ b/sqladmin/models.py @@ -51,6 +51,7 @@ Writer, get_object_identifier, get_primary_keys, + local_url_for, object_identifier_values, prettify_class_name, secure_filename, @@ -210,12 +211,6 @@ class UserAdmin(ModelView, model=User): # Internals pk_columns: ClassVar[Tuple[Column]] - session_maker: ClassVar[ # type: ignore[no-any-unimported] - Union[ - sessionmaker, - "async_sessionmaker", - ] - ] is_async: ClassVar[bool] = False is_model: ClassVar[bool] = True ajax_lookup_url: ClassVar[str] = "" @@ -703,7 +698,10 @@ class UserAdmin(ModelView, model=User): ``` """ - def __init__(self) -> None: + def __init__( + self, session_maker: Union[sessionmaker, "async_sessionmaker"] + ) -> None: + self.session_maker = session_maker self._mapper = inspect(self.model) self._prop_names = [attr.key for attr in self._mapper.attrs] self._relations = [ @@ -795,8 +793,8 @@ async def _run_query(self, stmt: ClauseElement) -> Any: def _url_for_delete(self, request: Request, obj: Any) -> str: pk = get_object_identifier(obj) query_params = urlencode({"pks": pk}) - url = request.url_for( - "admin:delete", identity=slugify_class_name(obj.__class__.__name__) + url = local_url_for( + request, "delete", identity=slugify_class_name(obj.__class__.__name__) ) return str(url) + "?" + query_params @@ -804,13 +802,14 @@ def _url_for_details_with_prop(self, request: Request, obj: Any, prop: str) -> U target = getattr(obj, prop, None) if target is None: return URL() - return self._build_url_for("admin:details", request, target) + return self._build_url_for("details", request, target) def _url_for_action(self, request: Request, action_name: str) -> str: - return str(request.url_for(f"admin:action-{self.identity}-{action_name}")) + return str(local_url_for(request, f"action-{self.identity}-{action_name}")) def _build_url_for(self, name: str, request: Request, obj: Any) -> URL: - return request.url_for( + return local_url_for( + request, name, identity=slugify_class_name(obj.__class__.__name__), pk=get_object_identifier(obj), diff --git a/sqladmin/templates/sqladmin/base.html b/sqladmin/templates/sqladmin/base.html index 2524648b..7b9ba0a5 100644 --- a/sqladmin/templates/sqladmin/base.html +++ b/sqladmin/templates/sqladmin/base.html @@ -4,12 +4,12 @@ - - - - - - + + + + + + {% if admin.favicon_url %} {% endif %} @@ -25,13 +25,13 @@ {% endblock %} {% endblock %} - - - - - - - + + + + + + + {% block tail %} {% endblock %} diff --git a/sqladmin/templates/sqladmin/create.html b/sqladmin/templates/sqladmin/create.html index e5557979..9234fd21 100644 --- a/sqladmin/templates/sqladmin/create.html +++ b/sqladmin/templates/sqladmin/create.html @@ -7,7 +7,7 @@

New {{ model_view.name }}

-
{% if error %} @@ -19,7 +19,7 @@

New {{ model_view.name }}

- + Cancel
diff --git a/sqladmin/templates/sqladmin/details.html b/sqladmin/templates/sqladmin/details.html index 8e1958a2..f1cb9e73 100644 --- a/sqladmin/templates/sqladmin/details.html +++ b/sqladmin/templates/sqladmin/details.html @@ -29,9 +29,9 @@

{% for elem, formatted_elem in zip(value, formatted_value) %} {% if model_view.show_compact_lists %} - ({{ formatted_elem }}) + ({{ formatted_elem }}) {% else %} - {{ formatted_elem }}
+ {{ formatted_elem }}
{% endif %} {% endfor %} @@ -50,7 +50,7 @@

- + Go Back
@@ -65,7 +65,7 @@

{% endif %} {% if model_view.can_edit %}
- + Edit
diff --git a/sqladmin/templates/sqladmin/edit.html b/sqladmin/templates/sqladmin/edit.html index c84507d5..60201461 100644 --- a/sqladmin/templates/sqladmin/edit.html +++ b/sqladmin/templates/sqladmin/edit.html @@ -7,7 +7,7 @@

Edit {{ model_view.name }}

-
{% if error %} @@ -19,7 +19,7 @@

Edit {{ model_view.name }}

- + Cancel
diff --git a/sqladmin/templates/sqladmin/layout.html b/sqladmin/templates/sqladmin/layout.html index ef739ebb..f885c15c 100644 --- a/sqladmin/templates/sqladmin/layout.html +++ b/sqladmin/templates/sqladmin/layout.html @@ -5,7 +5,7 @@
{% elif model_view.export_types | length == 1 %}
- Export @@ -35,7 +35,7 @@

{{ model_view.name_plural }}

{% endif %} {% if model_view.can_create %}
- + + New {{ model_view.name }}
@@ -54,7 +54,7 @@

{{ model_view.name_plural }}

{% if model_view.can_delete %} Delete selected items {% endif %} {% for custom_action, label in model_view._custom_actions_in_list.items() %} @@ -124,13 +124,13 @@

{{ model_view.name_plural }}

{% if model_view.can_view_details %} - {% endif %} {% if model_view.can_edit %} - @@ -152,9 +152,9 @@

{{ model_view.name_plural }}

{% for elem, formatted_elem in zip(value, formatted_value) %} {% if model_view.show_compact_lists %} - ({{ formatted_elem }}) + ({{ formatted_elem }}) {% else %} - {{ formatted_elem }}
+ {{ formatted_elem }}
{% endif %} {% endfor %} diff --git a/sqladmin/templates/sqladmin/login.html b/sqladmin/templates/sqladmin/login.html index dfb3e5a6..e38ae015 100644 --- a/sqladmin/templates/sqladmin/login.html +++ b/sqladmin/templates/sqladmin/login.html @@ -1,7 +1,7 @@ {% extends "sqladmin/base.html" %} {% block body %}
- +

Login to {{ admin.title }}

diff --git a/sqladmin/templating.py b/sqladmin/templating.py index cce68fde..57e36cde 100644 --- a/sqladmin/templating.py +++ b/sqladmin/templating.py @@ -9,6 +9,8 @@ from starlette.responses import HTMLResponse from starlette.types import Receive, Scope, Send +from sqladmin.helpers import local_url_for + class _TemplateResponse(HTMLResponse): def __init__( @@ -46,7 +48,7 @@ def __init__(self, directory: str) -> None: @jinja2.pass_context def url_for(context: dict, __name: str, **path_params: Any) -> URL: request: Request = context["request"] - return request.url_for(__name, **path_params) + return local_url_for(request, __name, **path_params) loader = jinja2.FileSystemLoader(directory) self.env = jinja2.Environment(loader=loader, autoescape=True, enable_async=True) diff --git a/tests/test_ajax.py b/tests/test_ajax.py index 1b8dd7ba..c610320e 100644 --- a/tests/test_ajax.py +++ b/tests/test_ajax.py @@ -222,10 +222,14 @@ async def test_ajax_response_limit(client: AsyncClient) -> None: async def test_create_ajax_loader_exceptions() -> None: with pytest.raises(ValueError): - create_ajax_loader(model_admin=AddressAdmin(), name="x", options={}) + create_ajax_loader( + model_admin=AddressAdmin(session_maker), name="x", options={}, + ) with pytest.raises(ValueError): - create_ajax_loader(model_admin=AddressAdmin(), name="user", options={}) + create_ajax_loader( + model_admin=AddressAdmin(session_maker), name="user", options={}, + ) async def test_create_page_template(client: AsyncClient) -> None: diff --git a/tests/test_application.py b/tests/test_application.py index f33eca58..234d95d8 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -105,7 +105,7 @@ class UserAdmin(ModelView, model=User): admin.add_view(UserAdmin) - @app.route("/{identity}", methods=["POST"]) + @admin.admin.route("/{identity}", methods=["POST"]) async def index(request: Request): obj = User(id=1) form_data = await request.form() @@ -114,16 +114,16 @@ async def index(request: Request): client = TestClient(app) - response = client.post("/user", data={"save": "Save"}) + response = client.post("/admin/user", data={"save": "Save"}) assert response.text == "http://testserver/admin/user/list" - response = client.post("/user", data={"save": "Save and continue editing"}) + response = client.post("/admin/user", data={"save": "Save and continue editing"}) assert response.text == "http://testserver/admin/user/edit/1" - response = client.post("/user", data={"save": "Save as new"}) + response = client.post("/admin/user", data={"save": "Save as new"}) assert response.text == "http://testserver/admin/user/edit/1" - response = client.post("/user", data={"save": "Save and add another"}) + response = client.post("/admin/user", data={"save": "Save and add another"}) assert response.text == "http://testserver/admin/user/create" diff --git a/tests/test_authentication.py b/tests/test_authentication.py index 3965b875..d1e1d941 100644 --- a/tests/test_authentication.py +++ b/tests/test_authentication.py @@ -11,6 +11,7 @@ from sqladmin import Admin, BaseView, action, expose from sqladmin.authentication import AuthenticationBackend +from sqladmin.helpers import local_url_for from sqladmin.models import ModelView from tests.common import sync_engine as engine @@ -39,7 +40,7 @@ async def logout(self, request: Request) -> bool: async def authenticate(self, request: Request) -> bool: if "token" not in request.session: - return RedirectResponse(request.url_for("admin:login"), status_code=302) + return RedirectResponse(local_url_for(request, "login"), status_code=302) return True diff --git a/tests/test_forms/test_forms.py b/tests/test_forms/test_forms.py index f2b1fbbf..3d964922 100644 --- a/tests/test_forms/test_forms.py +++ b/tests/test_forms/test_forms.py @@ -228,7 +228,7 @@ class MyForm(Form): class UserAdmin(ModelView, model=User): form = MyForm - form_type = await UserAdmin().scaffold_form() + form_type = await UserAdmin(session_maker).scaffold_form() form = form_type() assert isinstance(form, MyForm) assert len(form._fields) == 1 diff --git a/tests/test_menu.py b/tests/test_menu.py index eeecbbb7..2277f0bc 100644 --- a/tests/test_menu.py +++ b/tests/test_menu.py @@ -1,11 +1,13 @@ from sqlalchemy import Column, Integer, String -from sqlalchemy.orm import declarative_base +from sqlalchemy.orm import declarative_base, sessionmaker from starlette.requests import Request from sqladmin import ModelView from sqladmin._menu import CategoryMenu, ItemMenu, Menu, ViewMenu +from tests.common import sync_engine as engine Base = declarative_base() # type: ignore +session_maker = sessionmaker(bind=engine) class User(Base): @@ -48,7 +50,7 @@ def test_category_menu_is_active_when_child_is_active(): "path_params": {"identity": "user"}, } ) - user_menu = ViewMenu(view=UserAdmin(), name="user") + user_menu = ViewMenu(view=UserAdmin(session_maker), name="user") category_menu = CategoryMenu(name="Models") category_menu.add_child(user_menu) @@ -58,7 +60,7 @@ def test_category_menu_is_active_when_child_is_active(): def test_category_menu_is_not_active_when_no_child_is_active(): - user_menu = ViewMenu(view=UserAdmin(), name="user") + user_menu = ViewMenu(view=UserAdmin(session_maker), name="user") category_menu = CategoryMenu(name="Models") category_menu.add_child(user_menu) @@ -68,7 +70,7 @@ def test_category_menu_is_not_active_when_no_child_is_active(): def test_view_menu(): - item_menu = ViewMenu(view=UserAdmin(), name="view") + item_menu = ViewMenu(view=UserAdmin(session_maker), name="view") assert item_menu.display_name == "Users" assert item_menu.type_ == "View" diff --git a/tests/test_models.py b/tests/test_models.py index 47a695fa..6b0b7f92 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -146,21 +146,21 @@ def test_column_list_default() -> None: class UserAdmin(ModelView, model=User): pass - assert UserAdmin().get_list_columns() == ["id"] + assert UserAdmin(session_maker).get_list_columns() == ["id"] def test_column_list_by_model_columns() -> None: class UserAdmin(ModelView, model=User): column_list = [User.id, User.name] - assert UserAdmin().get_list_columns() == ["id", "name"] + assert UserAdmin(session_maker).get_list_columns() == ["id", "name"] def test_column_list_by_str_name() -> None: class AddressAdmin(ModelView, model=Address): column_list = ["id", "user_id"] - assert AddressAdmin().get_list_columns() == ["id", "user_id"] + assert AddressAdmin(session_maker).get_list_columns() == ["id", "user_id"] def test_column_filters() -> None: @@ -169,7 +169,7 @@ def test_column_filters() -> None: class UserAdmin(ModelView, model=User): column_filters = [filter] - all_filters = UserAdmin().get_filters() + all_filters = UserAdmin(session_maker).get_filters() assert len(all_filters) == 1 assert all_filters[0] == filter @@ -188,14 +188,18 @@ def test_column_exclude_list_by_str_name() -> None: class UserAdmin(ModelView, model=User): column_exclude_list = ["id"] - assert UserAdmin().get_list_columns() == ["addresses", "profile", "groups", "name"] + assert UserAdmin(session_maker).get_list_columns() == [ + "addresses", "profile", "groups", "name", + ] def test_column_exclude_list_by_model_column() -> None: class UserAdmin(ModelView, model=User): column_exclude_list = [User.id] - assert UserAdmin().get_list_columns() == ["addresses", "profile", "groups", "name"] + assert UserAdmin(session_maker).get_list_columns() == [ + "addresses", "profile", "groups", "name", + ] async def test_column_list_formatters() -> None: @@ -207,8 +211,10 @@ class UserAdmin(ModelView, model=User): user = User(id=1, name="Long Name") - assert await UserAdmin().get_list_value(user, "id") == (1, 2) - assert await UserAdmin().get_list_value(user, "name") == ("Long Name", "L") + assert await UserAdmin(session_maker).get_list_value(user, "id") == (1, 2) + assert await UserAdmin( + session_maker, + ).get_list_value(user, "name") == ("Long Name", "L") async def test_column_formatters_detail() -> None: @@ -220,8 +226,12 @@ class UserAdmin(ModelView, model=User): user = User(id=1, name="Long Name") - assert await UserAdmin().get_detail_value(user, "id") == (1, 2) - assert await UserAdmin().get_detail_value(user, "name") == ("Long Name", "L") + assert await UserAdmin( + session_maker, + ).get_detail_value(user, "id") == (1, 2) + assert await UserAdmin( + session_maker, + ).get_detail_value(user, "name") == ("Long Name", "L") async def test_column_formatters_default() -> None: @@ -230,11 +240,11 @@ class ProfileAdmin(ModelView, model=Profile): ... user = User(id=1, name="Long Name") profile = Profile(user=user, is_active=True) - assert await ProfileAdmin().get_list_value(profile, "is_active") == ( + assert await ProfileAdmin(session_maker).get_list_value(profile, "is_active") == ( True, Markup(""), ) - assert await ProfileAdmin().get_detail_value(profile, "is_active") == ( + assert await ProfileAdmin(session_maker).get_detail_value(profile, "is_active") == ( True, Markup(""), ) @@ -256,7 +266,7 @@ def test_column_details_list_default() -> None: class UserAdmin(ModelView, model=User): pass - assert UserAdmin().get_details_columns() == [ + assert UserAdmin(session_maker).get_details_columns() == [ "addresses", "profile", "groups", @@ -269,14 +279,14 @@ def test_column_details_list_by_model_column() -> None: class UserAdmin(ModelView, model=User): column_details_list = [User.name, User.id] - assert UserAdmin().get_details_columns() == ["name", "id"] + assert UserAdmin(session_maker).get_details_columns() == ["name", "id"] def test_column_details_exclude_list_by_model_column() -> None: class UserAdmin(ModelView, model=User): column_details_exclude_list = [User.id] - assert UserAdmin().get_details_columns() == [ + assert UserAdmin(session_maker).get_details_columns() == [ "addresses", "profile", "groups", @@ -288,7 +298,7 @@ def test_form_columns_default() -> None: class UserAdmin(ModelView, model=User): pass - assert UserAdmin().get_form_columns() == [ + assert UserAdmin(session_maker).get_form_columns() == [ "addresses", "profile", "groups", @@ -301,14 +311,16 @@ def test_form_columns_by_model_columns() -> None: class UserAdmin(ModelView, model=User): form_columns = [User.id, User.profile, User.name, User.addresses] - assert UserAdmin().get_form_columns() == ["id", "profile", "name", "addresses"] + assert UserAdmin(session_maker).get_form_columns() == [ + "id", "profile", "name", "addresses", + ] def test_form_columns_by_str_name() -> None: class AddressAdmin(ModelView, model=Address): form_columns = ["id", "user_id"] - assert AddressAdmin().get_form_columns() == ["id", "user_id"] + assert AddressAdmin(session_maker).get_form_columns() == ["id", "user_id"] def test_form_columns_both_include_and_exclude() -> None: @@ -325,47 +337,51 @@ def test_form_excluded_columns_by_str_name() -> None: class UserAdmin(ModelView, model=User): form_excluded_columns = ["id"] - assert UserAdmin().get_form_columns() == ["addresses", "profile", "groups", "name"] + assert UserAdmin(session_maker).get_form_columns() == [ + "addresses", "profile", "groups", "name", + ] def test_form_excluded_columns_by_model_column() -> None: class UserAdmin(ModelView, model=User): form_excluded_columns = [User.id] - assert UserAdmin().get_form_columns() == ["addresses", "profile", "groups", "name"] + assert UserAdmin(session_maker).get_form_columns() == [ + "addresses", "profile", "groups", "name", + ] def test_export_columns_default() -> None: class UserAdmin(ModelView, model=User): pass - assert UserAdmin().get_export_columns() == ["id"] + assert UserAdmin(session_maker).get_export_columns() == ["id"] def test_export_columns_default_to_list_columns() -> None: class UserAdmin(ModelView, model=User): column_list = [User.id, User.name] - assert UserAdmin().get_export_columns() == ["id", "name"] + assert UserAdmin(session_maker).get_export_columns() == ["id", "name"] class UserAdmin2(ModelView, model=User): column_list = [User.id] - assert UserAdmin2().get_export_columns() == ["id"] + assert UserAdmin2(session_maker).get_export_columns() == ["id"] def test_export_columns_by_model_columns() -> None: class UserAdmin(ModelView, model=User): column_export_list = [User.id, User.name] - assert UserAdmin().get_export_columns() == ["id", "name"] + assert UserAdmin(session_maker).get_export_columns() == ["id", "name"] def test_export_columns_by_str_name() -> None: class AddressAdmin(ModelView, model=Address): column_export_list = ["id", "user_id"] - assert AddressAdmin().get_export_columns() == ["id", "user_id"] + assert AddressAdmin(session_maker).get_export_columns() == ["id", "user_id"] def test_export_columns_both_include_and_exclude() -> None: @@ -384,7 +400,7 @@ def test_export_excluded_columns_by_str_name() -> None: class UserAdmin(ModelView, model=User): column_export_exclude_list = ["id"] - assert UserAdmin().get_export_columns() == [ + assert UserAdmin(session_maker).get_export_columns() == [ "addresses", "profile", "groups", @@ -396,7 +412,7 @@ def test_export_excluded_columns_by_model_column() -> None: class UserAdmin(ModelView, model=User): column_export_exclude_list = [User.id] - assert UserAdmin().get_export_columns() == [ + assert UserAdmin(session_maker).get_export_columns() == [ "addresses", "profile", "groups", @@ -417,22 +433,24 @@ class PostgresModel(Base): def test_model_default_sort() -> None: class UserAdmin(ModelView, model=User): ... - assert UserAdmin()._get_default_sort() == [("id", False)] + assert UserAdmin(session_maker)._get_default_sort() == [("id", False)] class UserAdmin(ModelView, model=User): column_default_sort = "name" - assert UserAdmin()._get_default_sort() == [("name", False)] + assert UserAdmin(session_maker)._get_default_sort() == [("name", False)] class UserAdmin(ModelView, model=User): column_default_sort = ("name", True) - assert UserAdmin()._get_default_sort() == [("name", True)] + assert UserAdmin(session_maker)._get_default_sort() == [("name", True)] class UserAdmin(ModelView, model=User): column_default_sort = [("name", True), ("id", False)] - assert UserAdmin()._get_default_sort() == [("name", True), ("id", False)] + assert UserAdmin(session_maker)._get_default_sort() == [ + ("name", True), ("id", False), + ] async def test_get_model_objects_uses_list_query() -> None: @@ -448,7 +466,7 @@ class UserAdmin(ModelView, model=User): def list_query(self, request: Request) -> Select: return super().list_query(request).filter(User.name.endswith("man")) - view = UserAdmin() + view = UserAdmin(session_maker) request = Request({"type": "http"}) assert len(await view.get_model_objects(request)) == 1 @@ -468,7 +486,7 @@ class UserAdmin(ModelView, model=User): async_engine = False session_maker = session_maker - view = UserAdmin() + view = UserAdmin(session_maker) request = Request({"type": "http", "path_params": {"pk": 123}}) user = await view.get_object_for_details(request) assert len(user.groups) == 2 @@ -496,7 +514,7 @@ def form_edit_query(self, request: Request) -> Select: .filter(Address.name == "bat cave") ) - view = UserAdmin() + view = UserAdmin(session_maker) class RequestObject(object): pass @@ -513,8 +531,12 @@ class AddressAdmin(ModelView, model=Address): column_list = "__all__" column_details_list = "__all__" - assert AddressAdmin().get_list_columns() == ["user", "id", "name", "user_id"] - assert AddressAdmin().get_details_columns() == ["user", "id", "name", "user_id"] + assert AddressAdmin(session_maker).get_list_columns() == [ + "user", "id", "name", "user_id", + ] + assert AddressAdmin(session_maker).get_details_columns() == [ + "user", "id", "name", "user_id", + ] async def test_get_prop_value() -> None: @@ -528,9 +550,13 @@ class ProfileAdmin(ModelView, model=Profile): session.add_all([user, address, profile]) session.commit() - assert await ProfileAdmin().get_prop_value(profile, "role") == "ADMIN" - assert await ProfileAdmin().get_prop_value(profile, "status") == "ACTIVE" - assert await ProfileAdmin().get_prop_value(profile, "user.name") == "admin" + assert await ProfileAdmin(session_maker).get_prop_value(profile, "role") == "ADMIN" + assert await ProfileAdmin( + session_maker, + ).get_prop_value(profile, "status") == "ACTIVE" + assert await ProfileAdmin( + session_maker, + ).get_prop_value(profile, "user.name") == "admin" async def test_model_property_in_columns() -> None: @@ -539,15 +565,21 @@ class UserAdmin(ModelView, model=User): user = User(id=1, name="batman") - assert UserAdmin().get_list_columns() == ["id", "name", "name_with_id"] - assert UserAdmin().get_details_columns() == [ + assert UserAdmin( + session_maker + ).get_list_columns() == ["id", "name", "name_with_id"] + assert UserAdmin( + session_maker + ).get_details_columns() == [ "addresses", "profile", "groups", "id", "name", ] - assert await UserAdmin().get_prop_value(user, "name_with_id") == "batman - 1" + assert await UserAdmin( + session_maker + ).get_prop_value(user, "name_with_id") == "batman - 1" def test_sort_query() -> None: @@ -556,15 +588,15 @@ class AddressAdmin(ModelView, model=Address): ... query = select(Address) request = Request({"type": "http", "query_string": "sortBy=id&sort=asc"}) - stmt = AddressAdmin().sort_query(query, request) + stmt = AddressAdmin(session_maker).sort_query(query, request) assert "ORDER BY addresses.id ASC" in str(stmt) request = Request({"type": "http", "query_string": b"sortBy=user.name&sort=desc"}) - stmt = AddressAdmin().sort_query(query, request) + stmt = AddressAdmin(session_maker).sort_query(query, request) assert "ORDER BY users.name DESC" in str(stmt) request = Request({"type": "http", "query_string": b"sortBy=user.profile.role"}) - stmt = AddressAdmin().sort_query(query, request) + stmt = AddressAdmin(session_maker).sort_query(query, request) assert "ORDER BY profiles.role ASC" in str(stmt) @@ -573,7 +605,7 @@ class AddressAdmin(ModelView, model=Address): ... request = Request({"type": "http"}) - stmt = AddressAdmin().count_query(request) + stmt = AddressAdmin(session_maker).count_query(request) assert "SELECT count(addresses.id) AS count_1" in str(stmt) @@ -592,7 +624,7 @@ class AddressAdmin(ModelView, model=Address): session_maker = multi_bind_session_maker request = Request({"type": "http"}) - count = await AddressAdmin().count(request) + count = await AddressAdmin(session_maker).count(request) assert count == 0 @@ -600,7 +632,7 @@ def test_search_query() -> None: class AddressAdmin(ModelView, model=Address): column_searchable_list = ["user.name", "user.profile.role"] - stmt = AddressAdmin().search_query(select(Address), "example") + stmt = AddressAdmin(session_maker).search_query(select(Address), "example") assert "lower(CAST(users.name AS VARCHAR))" in str(stmt) assert "lower(CAST(profiles.role AS VARCHAR))" in str(stmt) diff --git a/tests/test_models_action.py b/tests/test_models_action.py index 94245019..038e4f18 100644 --- a/tests/test_models_action.py +++ b/tests/test_models_action.py @@ -1,5 +1,5 @@ from typing import Any, Generator, List -from unittest.mock import Mock +from unittest.mock import Mock, patch import pytest from sqlalchemy import Column, Integer, String @@ -12,6 +12,7 @@ from sqladmin import Admin, ModelView from sqladmin.application import action from sqladmin.filters import AllUniqueStringValuesFilter, OperationColumnFilter +from sqladmin.helpers import local_url_for from tests.common import sync_engine as engine Base: Any = declarative_base() @@ -55,7 +56,7 @@ class RequestObject(object): obj_strs.append(repr(obj)) response = RedirectResponse( - request.url_for("admin:list", identity=self.identity) + local_url_for(request, "list", identity=self.identity) ) response.headers["X-Objs"] = ",".join(obj_strs) return response @@ -204,11 +205,10 @@ def test_model_action(client: TestClient) -> None: "list-confirm": "!List Confirm?!", } - request = Mock(Request) - request.url_for = Mock() - - admin.views[0]._url_for_action(request, "test") - request.url_for.assert_called_with("admin:action-user-test") + request = Mock() + with patch('sqladmin.models.local_url_for') as mock_func: + admin.views[0]._url_for_action(request, "test") + mock_func.assert_called_with(request, "action-user-test") with Session() as session: user1 = User() diff --git a/tests/test_pretty_export.py b/tests/test_pretty_export.py index 8b47e159..d84a07c0 100644 --- a/tests/test_pretty_export.py +++ b/tests/test_pretty_export.py @@ -59,11 +59,10 @@ async def _get_csv_content(self, response: StreamingResponse) -> str: async def test_get_export_row_values_basic(self): class UserAdmin(ModelView, model=User): column_list = ["id", "name", "email"] - session_maker = session_maker is_async = False user = User(id=1, name="John Doe", email="john@example.com", is_active=True) - model_view = UserAdmin() + model_view = UserAdmin(session_maker) column_names = ["id", "name", "email"] values = await PrettyExport._get_export_row_values( @@ -78,7 +77,6 @@ class UserAdmin(ModelView, model=User): async def test_get_export_row_values_with_custom_export_cell(self): class UserAdmin(ModelView, model=User): column_list = ["id", "name", "email"] - session_maker = session_maker is_async = False async def custom_export_cell( @@ -91,7 +89,7 @@ async def custom_export_cell( return None user = User(id=1, name="John Doe", email="john@example.com") - model_view = UserAdmin() + model_view = UserAdmin(session_maker) column_names = ["id", "name", "email"] values = await PrettyExport._get_export_row_values( @@ -106,11 +104,10 @@ async def custom_export_cell( async def test_get_export_row_values_with_base_export_cell(self): class UserAdmin(ModelView, model=User): column_list = ["id", "name", "is_active"] - session_maker = session_maker is_async = False user = User(id=1, name="John Doe", is_active=True) - model_view = UserAdmin() + model_view = UserAdmin(session_maker) column_names = ["id", "name", "is_active"] values = await PrettyExport._get_export_row_values( @@ -125,11 +122,10 @@ class UserAdmin(ModelView, model=User): async def test_get_export_row_values_with_none_values(self): class UserAdmin(ModelView, model=User): column_list = ["id", "name", "email"] - session_maker = session_maker is_async = False user = User(id=1, name="John Doe", email=None) - model_view = UserAdmin() + model_view = UserAdmin(session_maker) column_names = ["id", "name", "email"] values = await PrettyExport._get_export_row_values( @@ -144,12 +140,11 @@ class UserAdmin(ModelView, model=User): async def test_get_export_row_values_with_relationships(self): class AddressAdmin(ModelView, model=Address): column_list = ["id", "street", "user.name"] - session_maker = session_maker is_async = False user = User(id=1, name="John Doe") address = Address(id=1, street="123 Main St", user=user) - model_view = AddressAdmin() + model_view = AddressAdmin(session_maker) column_names = ["id", "street", "user.name"] values = await PrettyExport._get_export_row_values( @@ -164,14 +159,13 @@ class AddressAdmin(ModelView, model=Address): async def test_pretty_export_csv_basic(self): class UserAdmin(ModelView, model=User): column_list = ["id", "name", "email"] - session_maker = session_maker is_async = False users = [ User(id=1, name="John Doe", email="john@example.com"), User(id=2, name="Jane Smith", email="jane@example.com"), ] - model_view = UserAdmin() + model_view = UserAdmin(session_maker) response = await PrettyExport.pretty_export_csv(model_view, users) assert isinstance(response, StreamingResponse) @@ -197,13 +191,12 @@ class UserAdmin(ModelView, model=User): "name": "Full Name", "email": "Email Address", } - session_maker = session_maker is_async = False users = [ User(id=1, name="John Doe", email="john@example.com"), ] - model_view = UserAdmin() + model_view = UserAdmin(session_maker) response = await PrettyExport.pretty_export_csv(model_view, users) csv_content = await self._get_csv_content(response) @@ -219,13 +212,12 @@ class UserAdmin(ModelView, model=User): column_labels = { "name": "Full Name", } - session_maker = session_maker is_async = False users = [ User(id=1, name="John Doe", email="john@example.com"), ] - model_view = UserAdmin() + model_view = UserAdmin(session_maker) response = await PrettyExport.pretty_export_csv(model_view, users) csv_content = await self._get_csv_content(response) @@ -239,7 +231,6 @@ async def test_pretty_export_csv_with_custom_formatters(self): class UserAdmin(ModelView, model=User): column_list = ["id", "name", "is_active"] column_labels = {"id": "ID", "name": "Name", "is_active": "Active Status"} - session_maker = session_maker is_async = False async def custom_export_cell( @@ -253,7 +244,7 @@ async def custom_export_cell( User(id=1, name="John Doe", is_active=True), User(id=2, name="Jane Smith", is_active=False), ] - model_view = UserAdmin() + model_view = UserAdmin(session_maker) response = await PrettyExport.pretty_export_csv(model_view, users) @@ -268,11 +259,10 @@ async def custom_export_cell( async def test_pretty_export_csv_empty_data(self): class UserAdmin(ModelView, model=User): column_list = ["id", "name", "email"] - session_maker = session_maker is_async = False users = [] - model_view = UserAdmin() + model_view = UserAdmin(session_maker) response = await PrettyExport.pretty_export_csv(model_view, users) @@ -288,13 +278,12 @@ async def test_pretty_export_csv_with_export_columns(self): class UserAdmin(ModelView, model=User): column_list = ["id", "name", "email", "is_active"] column_export_list = ["name", "email"] - session_maker = session_maker is_async = False users = [ User(id=1, name="John Doe", email="john@example.com", is_active=True), ] - model_view = UserAdmin() + model_view = UserAdmin(session_maker) response = await PrettyExport.pretty_export_csv(model_view, users) csv_content = await self._get_csv_content(response) @@ -308,13 +297,12 @@ async def test_pretty_export_csv_integration_with_model_view_export(self): class UserAdmin(ModelView, model=User): column_list = ["id", "name", "email"] use_pretty_export = True - session_maker = session_maker is_async = False users = [ User(id=1, name="John Doe", email="john@example.com"), ] - model_view = UserAdmin() + model_view = UserAdmin(session_maker) response = await model_view.export_data(users, "csv") csv_content = await self._get_csv_content(response) @@ -329,14 +317,13 @@ class UserAdmin(ModelView, model=User): async def test_pretty_export_csv_filename_generation(self): class UserAdmin(ModelView, model=User): column_list = ["id", "name"] - session_maker = session_maker is_async = False def get_export_name(self, export_type: str) -> str: return f"test_export_with_special_chars!@#.{export_type}" users = [User(id=1, name="John Doe")] - model_view = UserAdmin() + model_view = UserAdmin(session_maker) response = await PrettyExport.pretty_export_csv(model_view, users) content_disposition = response.headers["Content-Disposition"] From 2987d55653c24ca679479e8195df8ef7e6a3d4c4 Mon Sep 17 00:00:00 2001 From: Maxim Dementyev Date: Tue, 7 Apr 2026 21:41:24 +0300 Subject: [PATCH 2/4] minor fixes --- sqladmin/helpers.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/sqladmin/helpers.py b/sqladmin/helpers.py index ecc398dc..4a38af89 100644 --- a/sqladmin/helpers.py +++ b/sqladmin/helpers.py @@ -329,13 +329,12 @@ def is_async_session_maker(session_maker: sessionmaker) -> bool: def local_url_for(request: Request, name: str, **kwargs: Any) -> URL: target_router = request.app.router - start_router = request.scope["router"] - if start_router == target_router: + start_router = request.scope.get("router") + if not start_router or start_router == target_router: return request.url_for(name, **kwargs) router_name = get_current_router_name(start_router, target_router) - return request.url_for( - f"{f'{router_name}:' if router_name else ''}{name}", **kwargs - ) + name_prefix = f'{router_name}:' if router_name else '' + return request.url_for(f"{name_prefix}{name}", **kwargs) def get_current_router_name(start_router: Any, target_router: Router) -> str | None: From 822d295559626456eed966e0108c6f7cd7b2015b Mon Sep 17 00:00:00 2001 From: Maxim Dementyev Date: Tue, 7 Apr 2026 21:41:46 +0300 Subject: [PATCH 3/4] add docstring --- sqladmin/helpers.py | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/sqladmin/helpers.py b/sqladmin/helpers.py index 4a38af89..9bc57529 100644 --- a/sqladmin/helpers.py +++ b/sqladmin/helpers.py @@ -328,6 +328,21 @@ def is_async_session_maker(session_maker: sessionmaker) -> bool: def local_url_for(request: Request, name: str, **kwargs: Any) -> URL: + """ + Generate a URL for the specified route, taking router hierarchy into account. + + If the current router matches the target router or is not defined, uses the standard + `request.url_for()` method. Otherwise, adds a router name prefix to the route name. + + Args: + request (Request): HTTP request object. + name (str): route name for URL generation. + **kwargs (Any): additional parameters for `request.url_for()` + (e.g., path parameters). + + Returns: + URL: generated URL for the specified route. + """ target_router = request.app.router start_router = request.scope.get("router") if not start_router or start_router == target_router: @@ -338,6 +353,20 @@ def local_url_for(request: Request, name: str, **kwargs: Any) -> URL: def get_current_router_name(start_router: Any, target_router: Router) -> str | None: + """ + Find the router name in the hierarchy that corresponds to the target router. + + Traverses the routes of the start router and its nested applications (of type `Mount`) + to find a match with the target router. + + Args: + start_router (Any): starting router from which the search begins. + target_router (Router): target router whose name needs to be found. + + Returns: + str | None: router name (possibly composite in the format `parent:child`) + if the target router is found. `None` if the router is not found in the hierarchy. + """ for router in getattr(start_router, "routes", []): if router == target_router: return router.name From 6594a237187be6db249fa13458b282a8fe5a621f Mon Sep 17 00:00:00 2001 From: Maxim Dementyev Date: Tue, 7 Apr 2026 22:15:03 +0300 Subject: [PATCH 4/4] fix lint --- sqladmin/helpers.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/sqladmin/helpers.py b/sqladmin/helpers.py index 9bc57529..325343d8 100644 --- a/sqladmin/helpers.py +++ b/sqladmin/helpers.py @@ -348,7 +348,7 @@ def local_url_for(request: Request, name: str, **kwargs: Any) -> URL: if not start_router or start_router == target_router: return request.url_for(name, **kwargs) router_name = get_current_router_name(start_router, target_router) - name_prefix = f'{router_name}:' if router_name else '' + name_prefix = f"{router_name}:" if router_name else "" return request.url_for(f"{name_prefix}{name}", **kwargs) @@ -356,8 +356,8 @@ def get_current_router_name(start_router: Any, target_router: Router) -> str | N """ Find the router name in the hierarchy that corresponds to the target router. - Traverses the routes of the start router and its nested applications (of type `Mount`) - to find a match with the target router. + Traverses the routes of the start router and its nested applications + (of type `Mount`) to find a match with the target router. Args: start_router (Any): starting router from which the search begins. @@ -365,7 +365,8 @@ def get_current_router_name(start_router: Any, target_router: Router) -> str | N Returns: str | None: router name (possibly composite in the format `parent:child`) - if the target router is found. `None` if the router is not found in the hierarchy. + if the target router is found.`None` if the router is not found + in the hierarchy. """ for router in getattr(start_router, "routes", []): if router == target_router: