👨‍🔬 Implement web backoffice using ASGI Admin

server/polar/admin/__init__.py

@@ -0,0 +1,15 @@
+from asgi_admin.views import AdminViewGroup, AdminViewIndex
+
+from .account import account_viewgroup
+from .organization import organization_viewgroup
+
+admin_viewgroup = AdminViewGroup(
+    index_view="index",
+    children=[
+        AdminViewIndex("/"),
+        organization_viewgroup,
+        account_viewgroup,
+    ],
+)
+
+__all__ = ["admin_viewgroup"]

server/polar/admin/account.py

@@ -0,0 +1,55 @@
+import functools
+from typing import Any
+
+from asgi_admin.integrations.sqlalchemy import RepositoryBase
+from asgi_admin.views import ModelViewEdit, ModelViewGroup, ModelViewList
+from sqlalchemy import Select, select
+
+from polar.models import Account
+
+from .base import get_repository
+
+
+class AccountRepository(RepositoryBase[Account]):
+    model = Account
+
+    def get_pk(self, item: Account) -> Any:
+        return item.id
+
+    def get_title(self, item: Account) -> str:
+        return str(item.id)
+
+    def get_base_select(self) -> Select[tuple[Account]]:
+        return select(Account).where(Account.deleted_at.is_(None))
+
+
+account_viewgroup = ModelViewGroup[Account](
+    "/accounts",
+    "accounts",
+    title="Accounts",
+    get_repository=functools.partial(get_repository, AccountRepository),
+    index_view="list",
+    children=[
+        ModelViewList[Account](
+            path="/",
+            name="list",
+            title="List",
+            fields=(
+                ("id", "ID"),
+                ("status", "Status"),
+                ("account_type", "Type"),
+                ("created_at", "Created At"),
+                ("next_review_threshold", "Next Review Threshold"),
+            ),
+            query_fields=("id", "status", "account_type"),
+        ),
+        ModelViewEdit[Account](
+            path="/{pk}",
+            name="edit",
+            title="Edit",
+            fields=tuple(),
+        ),
+    ],
+)
+
+__all__ = ["account_viewgroup"]

server/polar/admin/base.py

@@ -0,0 +1,16 @@
+from collections.abc import AsyncIterator
+
+from asgi_admin.integrations.sqlalchemy import RepositoryBase
+from asgi_admin.repository import Model
+from starlette.requests import Request
+
+
+async def get_repository(
+    repository_class: type[RepositoryBase[Model]], request: Request
+) -> AsyncIterator[RepositoryBase[Model]]:
+    async with request.state.async_sessionmaker() as session:
+        yield repository_class(session)
+        await session.commit()
+
+
+__all__ = ["get_repository"]

server/polar/admin/organization.py

@@ -0,0 +1,98 @@
+import functools
+from typing import Any
+
+from asgi_admin.integrations.sqlalchemy import RepositoryBase
+from asgi_admin.views import ModelViewEdit, ModelViewGroup, ModelViewList
+from sqlalchemy import Select, select
+from starlette.requests import Request
+from wtforms import Form, StringField, validators
+
+from polar.models import Organization
+
+from .base import get_repository
+
+
+class OrganizationRepository(RepositoryBase[Organization]):
+    model = Organization
+
+    def get_pk(self, item: Organization) -> Any:
+        return item.id
+
+    def get_title(self, item: Organization) -> str:
+        return item.name
+
+    def get_base_select(self) -> Select[tuple[Organization]]:
+        return select(Organization).where(Organization.deleted_at.is_(None))
+
+    async def get_by_slug(self, slug: str) -> Organization | None:
+        statement = self.get_base_select().where(Organization.slug == slug)
+        return await self.get_one_or_none(statement)
+
+
+async def validate_slug(
+    request: Request,
+    repository: OrganizationRepository,
+    organization: Organization,
+    form: Form,
+) -> bool:
+    if "slug" in form.data:
+        existing = await repository.get_by_slug(form.data["slug"])
+        if existing is not None and existing.id != organization.id:
+            form.slug.errors.append("Slug already exists.")
+            return False
+    return True
+
+
+organization_viewgroup = ModelViewGroup[Organization](
+    "/organizations",
+    "organizations",
+    title="Organizations",
+    get_repository=functools.partial(get_repository, OrganizationRepository),
+    index_view="list",
+    children=[
+        ModelViewList[Organization](
+            path="/",
+            name="list",
+            title="List",
+            fields=(
+                ("id", "ID"),
+                ("name", "Name"),
+                ("slug", "Slug"),
+                ("created_at", "Created At"),
+            ),
+            query_fields=("id", "name", "slug"),
+            details_view_name="edit",
+        ),
+        ModelViewEdit[Organization](
+            path="/{pk}",
+            name="edit",
+            title="Edit",
+            fields=(
+                (
+                    "name",
+                    StringField(
+                        "Name",
+                        validators=[
+                            validators.InputRequired(),
+                            validators.Length(min=3),
+                        ],
+                    ),
+                ),
+                (
+                    "slug",
+                    StringField(
+                        "Slug",
+                        filters=[str.lower],
+                        validators=[
+                            validators.InputRequired(),
+                            validators.Length(min=3),
+                        ],
+                    ),
+                ),
+            ),
+            async_validators=[validate_slug],
+        ),
+    ],
+)
+
+__all__ = ["organization_viewgroup"]

server/polar/app.py

@@ -7,6 +7,7 @@
 from fastapi.routing import APIRoute
 
 from polar import receivers, worker  # noqa
+from polar.admin import admin_viewgroup
 from polar.api import router
 from polar.checkout import ip_geolocation
 from polar.config import settings
@@ -161,6 +162,9 @@ def create_app() -> FastAPI:
     # /healthz
     app.include_router(health_router)
 
+    # /admin
+    app.mount("/admin", admin_viewgroup.route)
+
     app.include_router(router)
     document_webhooks(app)
 

server/pyproject.toml

@@ -49,6 +49,7 @@ dependencies = [
   "apscheduler>=3.10.4",
   "supervisor>=4.2.5",
   "plain-client>=0.0.1",
+  "asgi-admin",
 ]
 
 [dependency-groups]
@@ -147,3 +148,6 @@ init_forbid_extra = true
 init_typed = true
 warn_required_dynamic_aliases = true
 warn_untyped_fields = true
+
+[tool.uv.sources]
+asgi-admin = { git = "https://github.com/polarsource/asgi-admin", rev = "b6d7b8be89a13cf6bbd876ca895cbab25ec5fda8" }