README.md

@@ -57,3 +57,7 @@ uvicorn webfinger:app --port 7033
 ### Post-run
 
 proxy `/.well-known/webfinger` to localhost:7033
+
+## Development
+
+`pdm run uvicorn webfinger:app --reload`

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "webfinger"
-version = "1"
+version = "2"
 description = "Simple WebFinger server that returns static resources. Written with Python and FastAPI."
 authors = [
     {name = "a", email = "a@trwnh.com"},

webfinger/__init__.py

@@ -1,62 +1,32 @@
 from fastapi import FastAPI
-from fastapi.responses import PlainTextResponse, JSONResponse
+import webfinger.lookup
-from fastapi.params import Query
-import json
-from pathlib import Path
-from os import environ as env
 
-app = FastAPI()
+app = FastAPI(
-
+	title = "webfinger"
-@app.get("/")
-def lookup(
-	resource: str = Query(None),
-	rel: list[str] = Query(None)
-	):
-	"""
-	Respond to a WebFinger query.
-	"""
-
-	# Basic info as a hint, if no resource given
-	if not resource:
-		return PlainTextResponse(
-			content="Query ?resource= for more information.",
-			status_code=400,
 )
+app.include_router(webfinger.lookup.router)
 
-	# otherwise, load that resource and return its data
+from fastapi.openapi.utils import generate_operation_id
-	dir = env.get("RESOURCE_DIR") or "resource"
+from fastapi.routing import APIRoute
-	filename = f"{dir}/{resource}.json"
-	path = Path(filename)
 
-	# TODO redirect to upstream webfinger server(s) if resource not found locally
+def remove_422s(app: FastAPI) -> None:
-	#       and if there is an upstream to try
+    openapi_schema = app.openapi()
-	if not path.is_file():
+    operation_ids_to_update: set[str] = set()
-		return PlainTextResponse(
+    for route in app.routes:
-			content=f"{resource} not found",
+        if not isinstance(route, APIRoute):
-			status_code=404,
+            continue
-		)
+        methods = route.methods or ["GET"]
+        if getattr(route.endpoint, "__remove_422__", None):
+            for method in methods:
+                operation_ids_to_update.add(generate_operation_id(route=route, method=method))
+    paths = openapi_schema["paths"]
+    for _, operations in paths.items():
+        for method, metadata in operations.items():
+            operation_id = metadata.get("operationId")
+            if operation_id in operation_ids_to_update:
+                metadata["responses"].pop("422", None)
 
-	# open the local file
+remove_422s(app)
-	with open(filename, "r") as file:
-		data: dict = json.loads(file.read())
-
-	# filter links if rel is provided
-	links = [link for link in data.get('links', [])]
-	if rel:
-		links = [link for link in links if link.get('rel', '') in rel]
-
-	# construct json response
-	content = {
-		"subject": resource,
-		"aliases": [alias for alias in data.get('aliases')],
-		"properties": data.get('properties'),
-		"links": links,
-	}
-	content = {k: v for k, v in content.items() if v} # remove null values
-	headers = {
-		"Content-Type": "application/jrd+json"
-	}
-	return JSONResponse(content=content, headers=headers)
 
 if __name__ == "__main__":
 	import uvicorn

webfinger/lookup.py

@@ -0,0 +1,169 @@
+from fastapi import APIRouter, Response
+from fastapi.responses import PlainTextResponse, JSONResponse
+from fastapi.params import Query
+import json
+from pathlib import Path
+from os import environ as env
+from pydantic import BaseModel
+
+
+router = APIRouter(
+	prefix="/.well-known/webfinger"
+)
+
+
+## String literals
+
+MEDIA_TYPE = "application/jrd+json"
+
+EXAMPLE_RESOURCE = r'''
+{
+	"aliases": ["https://trwnh.com/actors/7057bc10-db1c-4ebe-9e00-22cf04be4e5e", "https://trwnh.com/~a"],
+	"links": [
+		{
+			"rel": "self",
+			"type": "application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"",
+			"href": "https://trwnh.com/actors/7057bc10-db1c-4ebe-9e00-22cf04be4e5e"
+		},
+		{
+			"rel": "https://webfinger.net/rel/profile-page/",
+			"type": "text/html",
+			"href": "https://trwnh.com/~a"
+		}
+	]
+}
+'''
+
+
+## Pydantic models for OpenAPI schema
+
+class Link(BaseModel):
+	rel: str
+	type: str | None
+	href: str | None
+	titles: str | None
+	properties: str | None
+
+	class Config:
+		schema_extra = {
+			"example": '''{
+			"rel": "https://webfinger.net/rel/profile-page/",
+			"type": "text/html",
+			"href": "https://trwnh.com/~a"
+		}'''
+		}
+
+class Resource(BaseModel):
+	subject: str
+	aliases: list[str] | None
+	links: list[Link] | None
+
+	class Config:
+		schema_extra = {
+			"example": EXAMPLE_RESOURCE
+		}
+
+class WebfingerResponse(Response):
+	media_type = MEDIA_TYPE
+
+
+## Example responses for OpenAPI schema
+
+RESPONSES = {
+	400: {
+		"description": "Bad Request. No resource was provided.",
+		"content": {
+			"text/plain": {
+				"example": "Query ?resource= for more information."
+			}
+		},
+	},
+	404: {
+		"description": "Not Found. Could not find a document for the provided resource.",
+		"content": {
+			"text/plain": {
+				"example": "{resource} not found"
+			}
+		},
+	},
+	200: {
+		"description": "The resource you requested was found, and has the returned resource document.",
+		"model": Resource,
+		"content": {
+			MEDIA_TYPE: {
+				"example": EXAMPLE_RESOURCE
+			}
+		}
+	},
+}
+
+
+## The lookup method as described in RFC7033 
+
+def remove_422(func):
+	func.__remove_422__ = True
+	return func
+
+@router.get(
+	"",
+	summary = "Lookup a Webfinger resource",
+	description = "Query the Webfinger service for a resource URI and obtain its associated resource document",
+	tags = ["RFC7033"],
+	response_model = Resource,
+	response_class = WebfingerResponse,
+	responses = {**RESPONSES},
+
+)
+@remove_422
+async def lookup(
+	resource: str = Query(
+		None,
+		description = "The subject whose document you are looking up. If not provided, you will get a 400 Bad Request."
+		),
+	rel: list[str] = Query(
+		None,
+		description = "If provided, filter the links array for this rel-value only. This parameter may be included multiple times."
+		)
+	):
+	"""
+	Respond to a WebFinger query.
+	"""
+
+	# Basic info as a hint, if no resource given
+	if not resource:
+		return PlainTextResponse(
+			content="Query ?resource= for more information.",
+			status_code=400,
+		)
+	
+	# otherwise, load that resource and return its data
+	dir = env.get("RESOURCE_DIR") or "resource"
+	filename = f"{dir}/{resource}.json"
+	path = Path(filename)
+
+	# TODO redirect to upstream webfinger server(s) if resource not found locally
+	#       and if there is an upstream to try
+	if not path.is_file():
+		return PlainTextResponse(
+			content=f"{resource} not found",
+			status_code=404,
+		)
+
+	# open the local file
+	with open(filename, "r") as file:
+		data: dict = json.loads(file.read())
+
+	# filter links if rel is provided
+	links = [link for link in data.get('links', [])]
+	if rel:
+		links = [link for link in links if link.get('rel', '') in rel]
+
+	# construct json response
+	content = {
+		"subject": resource,
+		"aliases": [alias for alias in data.get('aliases')],
+		"properties": data.get('properties'),
+		"links": links,
+	}
+	content = {k: v for k, v in content.items() if v} # remove null values
+	return JSONResponse(content=content, media_type=MEDIA_TYPE)