-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy path_admin.py
More file actions
138 lines (108 loc) · 4.77 KB
/
Copy path_admin.py
File metadata and controls
138 lines (108 loc) · 4.77 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
"""Admin orchestrator — mount under HawkAPI, register resources."""
from __future__ import annotations
import logging
import warnings
from collections.abc import Awaitable, Callable
from typing import Any
from ._resource import ModelResource
from ._routes import delete, detail, edit_form, index, list_resource, save
_logger = logging.getLogger("hawkapi_admin")
AuthCallable = Callable[[Any], Awaitable[None]]
async def _deny_all_auth(_request: Any) -> None:
"""Default fail-closed auth: deny every request when no auth is configured."""
from hawkapi import HTTPException
raise HTTPException(401, detail="Admin authentication is not configured")
class Admin:
"""Top-level admin object. Build one per HawkAPI app and register resources."""
def __init__(
self,
*,
title: str = "Admin",
url_prefix: str = "/admin",
auth: AuthCallable | None = None,
csrf_enabled: bool = True,
) -> None:
self.title = title
self.url_prefix = url_prefix.rstrip("/") or "/admin"
self.resources: dict[str, ModelResource] = {}
self.csrf_enabled: bool = csrf_enabled
if auth is None:
warnings.warn(
"hawkapi-admin: no auth configured; all admin endpoints will deny "
"access (fail-closed). Pass an `auth` callable to enable the panel.",
UserWarning,
stacklevel=2,
)
self.auth: AuthCallable = _deny_all_auth
else:
self.auth = auth
def register(self, resource: ModelResource | type[Any], **kwargs: Any) -> ModelResource:
"""Register ``resource`` with the admin. Accepts a SQLAlchemy model class
directly for the simple case::
admin.register(User)
or a fully customized :class:`ModelResource`::
admin.register(ModelResource(model=User, list_display=("id", "email"),
list_search=("email",)))
"""
if not isinstance(resource, ModelResource):
resource = ModelResource(model=resource, **kwargs)
if resource.name in self.resources:
raise ValueError(f"resource {resource.name!r} already registered")
self.resources[resource.name] = resource
return resource
async def _check_auth(self, request: Any) -> None:
await self.auth(request)
def attach(self, app: Any) -> None:
"""Wire the admin routes onto ``app``."""
prefix = self.url_prefix
admin = self
if self.auth is _deny_all_auth:
_logger.warning(
"hawkapi-admin: no auth configured; all admin endpoints deny access (fail-closed)"
)
async def _index(request: Any) -> Any:
await admin._check_auth(request)
return await index(request, admin=admin)
async def _list(request: Any) -> Any:
await admin._check_auth(request)
return await list_resource(request, admin=admin)
async def _detail(request: Any) -> Any:
await admin._check_auth(request)
return await detail(request, admin=admin)
async def _new_form(request: Any) -> Any:
await admin._check_auth(request)
return await edit_form(request, admin=admin)
async def _edit_form(request: Any) -> Any:
await admin._check_auth(request)
return await edit_form(request, admin=admin)
async def _save_new(request: Any) -> Any:
await admin._check_auth(request)
return await save(request, admin=admin)
async def _save_existing(request: Any) -> Any:
await admin._check_auth(request)
return await save(request, admin=admin)
async def _delete(request: Any) -> Any:
await admin._check_auth(request)
return await delete(request, admin=admin)
app.get(prefix)(_index)
app.get(f"{prefix}/{{resource}}")(_list)
app.get(f"{prefix}/{{resource}}/new")(_new_form)
app.post(f"{prefix}/{{resource}}/new")(_save_new)
app.get(f"{prefix}/{{resource}}/{{pk}}")(_detail)
app.get(f"{prefix}/{{resource}}/{{pk}}/edit")(_edit_form)
app.post(f"{prefix}/{{resource}}/{{pk}}/edit")(_save_existing)
app.post(f"{prefix}/{{resource}}/{{pk}}/delete")(_delete)
def init_admin(
app: Any,
*,
admin: Admin | None = None,
title: str = "Admin",
url_prefix: str = "/admin",
auth: AuthCallable | None = None,
csrf_enabled: bool = True,
) -> Admin:
"""Create an :class:`Admin`, attach it to ``app``, and return it for further registrations."""
admin = admin or Admin(title=title, url_prefix=url_prefix, auth=auth, csrf_enabled=csrf_enabled)
admin.attach(app)
return admin
__all__ = ["Admin", "init_admin"]