BEgin adding admin controls

account/config.py

@@ -40,6 +40,7 @@ class Settings(BaseModel):
     # reCaptcha
     recaptcha_secret_key: str = config("RECAPTCHA_SECRET_KEY", default="")
 
+    admin_root: str = config("ADMIN_ROOT", default="")
     testing: bool = config("TESTING", default=False, cast=bool)
 
 

account/main.py

@@ -2,7 +2,13 @@
 
 from . import routes
 from .app import app
+from .config import CONFIG
 
 
 for router in routes.ROUTERS:
     app.include_router(router)
+
+if CONFIG.admin_root:
+    from .routes import admin
+
+    app.include_router(admin.router)

account/models/helpers.py

@@ -4,6 +4,8 @@
 from typing import Any
 
 from bson.objectid import ObjectId, InvalidId
+# from pydantic_core import core_schema
+# from pydantic import GetCoreSchemaHandler
 
 
 class ObjectIdStr(str):
@@ -21,3 +23,16 @@ def validate(cls, value: Any, _: Any) -> str:
         except InvalidId as exc:
             raise ValueError("Not a valid ObjectId") from exc
         return str(value)
+
+    # Pydantic v2 -> v3 requirement
+    # Necessary to allow OpenAPI spec to build
+    # @classmethod
+    # def __get_pydantic_core_schema__(
+    #     cls, source: type[Any], handler: GetCoreSchemaHandler
+    # ) -> core_schema.CoreSchema:
+    #     assert issubclass(source, ObjectIdStr)
+    #     return core_schema.json_or_python_schema(
+    #         json_schema=core_schema.str_schema(),
+    #         python_schema=core_schema.is_instance_schema(str),
+    #         serialization=core_schema.plain_serializer_function_ser_schema(lambda c: c),
+    #     )

account/models/user.py

@@ -104,6 +104,7 @@ class UserOut(UserUpdate):
     allow_overage: bool = False
     subscribed: bool = False
     disabled: bool = False
+    is_admin: bool = False
 
 
 class User(Document, UserOut):

account/models/util.py

@@ -0,0 +1,9 @@
+"""Other output models."""
+
+from pydantic import BaseModel
+
+
+class JustUrl(BaseModel):
+    """Just returns a URL."""
+
+    url: str

account/routes/admin/__init__.py

@@ -0,0 +1,10 @@
+"""Admin routes."""
+
+from fastapi import APIRouter
+from account.config import CONFIG
+from . import stripe, token
+
+router = APIRouter(prefix=f"/{CONFIG.admin_root}", include_in_schema=False)
+
+# router.include_router(stripe.router)
+router.include_router(token.router)

account/routes/admin/stripe.py

@@ -0,0 +1,19 @@
+"""Billing admin routes."""
+
+from fastapi import APIRouter, Depends, HTTPException
+
+from account.models.user import User
+from account.models.util import JustUrl
+from account.util.current_user import admin_user, embedded_user
+from account.util.stripe import get_portal_session
+
+router = APIRouter(prefix="/stripe")
+
+
+@router.get("/portal", dependencies=[Depends(admin_user)])
+async def get_billing_url(user: User = Depends(embedded_user)) -> JustUrl:
+    """Return the Stripe account portal for another user."""
+    if not (user.stripe and user.stripe.customer_id):
+        raise HTTPException(400, "No stripe fields available")
+    session = get_portal_session(user)
+    return JustUrl(url=session.url)

account/routes/admin/token.py

@@ -0,0 +1,18 @@
+"""Token admin routes."""
+
+from fastapi import APIRouter, Depends
+
+from account.models.token import AllTokenUsageOut
+from account.models.user import User
+from account.util.current_user import admin_user, embedded_user
+from account.util.token import token_usage_for
+
+router = APIRouter(prefix="/token")
+
+
+@router.post("/history", dependencies=[Depends(admin_user)])
+async def get_all_history(
+    days: int = 30, user: User = Depends(embedded_user)
+) -> list[AllTokenUsageOut]:
+    """Return all recent token history for another user."""
+    return await token_usage_for(user, days)

account/routes/stripe.py

@@ -5,6 +5,7 @@
 from stripe import SignatureVerificationError
 
 from account.models.user import User, UserOut
+from account.models.util import JustUrl
 from account.util.current_user import current_user
 from account.util.stripe import (
     get_event,
@@ -70,9 +71,9 @@ async def stripe_fulfill(
 
 
 @router.get("/portal")
-async def customer_portal(user: User = Depends(current_user)) -> dict[str, str] | None:
+async def customer_portal(user: User = Depends(current_user)) -> JustUrl:
     """Return the user's Stripe account portal URL."""
     if not (user.stripe and user.stripe.customer_id):
-        return None
+        raise HTTPException(400, "No stripe fields available")
     session = get_portal_session(user)
-    return {"url": session.url}
+    return JustUrl(url=session.url)

account/routes/token.py

@@ -1,7 +1,6 @@
 """Token management router."""
 
 from datetime import datetime, timedelta, UTC
-from typing import Any
 
 from bson.objectid import ObjectId
 from fastapi import APIRouter, Depends, HTTPException, Response
@@ -15,6 +14,7 @@
 )
 from account.models.user import User, UserToken
 from account.util.current_user import current_user
+from account.util.token import token_usage_for
 
 router = APIRouter(prefix="/token", tags=["Token"])
 
@@ -34,34 +34,12 @@ async def new_token(user: User = Depends(current_user)) -> UserToken:
     return token
 
 
-@router.get("/history", response_model=list[AllTokenUsageOut])
+@router.get("/history")
 async def get_all_history(
     days: int = 30, user: User = Depends(current_user)
-) -> list[dict[str, Any]]:
+) -> list[AllTokenUsageOut]:
     """Return all recent token history."""
-    days_since = datetime.now(tz=UTC) - timedelta(days=days)
-    data = (
-        await TokenUsage.find(
-            TokenUsage.user_id == ObjectId(user.id),
-            TokenUsage.date >= days_since,
-        )
-        .aggregate(
-            [
-                {"$project": {"_id": 0, "date": 1, "count": 1, "token_id": 1}},
-                {
-                    "$group": {
-                        "_id": "$token_id",
-                        "days": {"$push": {"date": "$date", "count": "$count"}},
-                    }
-                },
-            ]
-        )
-        .to_list()
-    )
-    for i, item in enumerate(data):
-        data[i]["token_id"] = item["_id"]
-        del data[i]["_id"]
-    return data
+    return await token_usage_for(user, days)
 
 
 @router.get("/{value}", response_model=Token)
@@ -81,7 +59,7 @@ async def update_token(
     i, token = user.get_token(value)
     if token is None:
         raise HTTPException(404, f"Token with value {value} does not exist")
-    token = token.copy(update=update.dict(exclude_unset=True))
+    token = token.model_copy(update=update.model_dump(exclude_unset=True))
     user.tokens[i] = token
     await user.save()
     return token

account/util/current_user.py

@@ -1,6 +1,6 @@
 """Current user dependency."""
 
-from fastapi import HTTPException, Security
+from fastapi import Body, HTTPException, Security
 from fastapi_jwt import JwtAuthorizationCredentials
 
 from account.models.user import User
@@ -17,3 +17,23 @@ async def current_user(
     if user is None:
         raise HTTPException(404, "Authorized user could not be found")
     return user
+
+
+async def embedded_user(email: str = Body(..., embed=True)) -> User:
+    """Return a user from an embedded email."""
+    if not email:
+        raise HTTPException(401, "No user email found")
+    user = await User.by_email(email)
+    if user is None:
+        raise HTTPException(404, "Embedded user could not be found")
+    return user
+
+
+async def admin_user(
+    auth: JwtAuthorizationCredentials = Security(access_security),
+) -> User:
+    """Return the current admin user."""
+    user = await current_user(auth)
+    if not user.is_admin:
+        raise HTTPException(403, "Not allowed to access resource")
+    return user

account/util/token.py

@@ -0,0 +1,35 @@
+"""Shared token utilities."""
+
+from datetime import datetime, timedelta, UTC
+
+from bson.objectid import ObjectId
+
+from account.models.token import AllTokenUsageOut, TokenUsage
+from account.models.user import User
+
+
+async def token_usage_for(user: User, days: int) -> list[AllTokenUsageOut]:
+    """Get recent token history for a user."""
+    days_since = datetime.now(tz=UTC) - timedelta(days=days)
+    data = (
+        await TokenUsage.find(
+            TokenUsage.user_id == ObjectId(user.id),
+            TokenUsage.date >= days_since,
+        )
+        .aggregate(
+            [
+                {"$project": {"_id": 0, "date": 1, "count": 1, "token_id": 1}},
+                {
+                    "$group": {
+                        "_id": "$token_id",
+                        "days": {"$push": {"date": "$date", "count": "$count"}},
+                    }
+                },
+            ]
+        )
+        .to_list()
+    )
+    for i, item in enumerate(data):
+        data[i]["token_id"] = item["_id"]
+        del data[i]["_id"]
+    return [AllTokenUsageOut.model_validate(d) for d in data]

tests/conftest.py

@@ -6,7 +6,7 @@
 from asgi_lifespan import LifespanManager
 from decouple import config
 from fastapi import FastAPI
-from httpx import AsyncClient
+from httpx import AsyncClient, ASGITransport
 
 from account.config import CONFIG
 
@@ -29,7 +29,9 @@ async def clear_database(server: FastAPI) -> None:
 async def client() -> AsyncIterator[AsyncClient]:
     """Async server client that handles lifespan and teardown."""
     async with LifespanManager(app):
-        async with AsyncClient(app=app, base_url="http://test") as _client:
+        async with AsyncClient(
+            transport=ASGITransport(app=app), base_url="http://test"
+        ) as _client:
             try:
                 yield _client
             except Exception as exc: