Use pydantic on ResourceServer
All checks were successful
/ build (push) Successful in 5s
/ test (push) Successful in 5s

This commit is contained in:
phil 2025-02-13 12:23:18 +01:00
parent 0464047f8a
commit 381ce1ebc1
3 changed files with 16 additions and 10 deletions

View file

@ -58,7 +58,7 @@ async def lifespan(app: FastAPI):
try:
await provider.get_info()
except NoPublicKey:
logger.warn(f"Disable {provider.id}: public key not found")
logger.warning(f"Disable {provider.id}: public key not found")
del providers[provider.id]
yield
@ -300,7 +300,6 @@ async def refresh(
await update_token(provider.id, new_token)
return RedirectResponse(url=request.url_for("home"))
# Snippet for running standalone
# Mostly useful for the --version option,
# as running with uvicorn is easy and provides better flexibility, eg.

View file

@ -17,20 +17,20 @@ class ProcessError(Exception):
pass
class ResourceProvider:
name: str
class ResourceProvider(BaseModel):
scope_required: str | None = None
default_resource_id: str | None = None
def __init__(self, name: str):
self.name = name
super().__init__()
self.__name__ = name
async def process(self, user: User, resource_id: str | None = None) -> ProcessResult:
logger.warning(f"{self.name} should define a process method")
logger.warning(f"{self.__name__} should define a process method")
return ProcessResult()
class ResourceRegistry:
class ResourceRegistry(BaseModel):
resource_providers: dict[str, ResourceProvider] = {}
def make_registry(self):

View file

@ -1,5 +1,4 @@
from datetime import datetime
from typing import Annotated
from typing import Annotated, Any
import logging
from authlib.oauth2.auth import OAuth2Token
@ -21,7 +20,7 @@ from oidc_test.auth.utils import (
from oidc_test.auth_providers import providers
from oidc_test.settings import settings
from oidc_test.models import User
from oidc_test.registry import ProcessError, ProcessResult, registry
from oidc_test.registry import ProcessError, ProcessResult, ResourceProvider, registry
logger = logging.getLogger("oidc-test")
@ -48,6 +47,14 @@ resource_server.add_middleware(
# Routes for RBAC based tests
@resource_server.get("/")
async def resources() -> dict[str, dict[str, Any]]:
return {
"internal": {},
"plugins": registry.resource_providers
}
@resource_server.get("/public")
async def public() -> dict:
return {"msg": "Not protected"}