Enhance documentation/API documentation and examples

API/auth/__init__.py

@@ -1,13 +1,18 @@
 from enum import Enum
 from typing import Union
 
-from fastapi import Depends, Header, HTTPException
+from fastapi.security import APIKeyHeader
+from fastapi import Depends, HTTPException
 from osm_login_python.core import Auth
 from pydantic import BaseModel, Field
 
 from src.app import Users
 from src.config import get_oauth_credentials
 
+API_Access_Token = APIKeyHeader(
+    name="Access_Token", description="Access Token to Authorize User", auto_error=False
+)
+
 
 class UserRole(Enum):
     ADMIN = 1
@@ -21,6 +26,16 @@ class AuthUser(BaseModel):
     img_url: Union[str, None]
     role: UserRole = Field(default=UserRole.GUEST.value)
 
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "id": "123",
+                "username": "HOT Team",
+                "img_url": "https://hotosm/image.jpg",
+                "role": UserRole.GUEST.value,
+            }
+        }
+
 
 osm_auth = Auth(*get_oauth_credentials())
 
@@ -34,7 +49,7 @@ def get_user_from_db(osm_id: int):
 def get_osm_auth_user(access_token):
     try:
         user = AuthUser(**osm_auth.deserialize_access_token(access_token))
-    except Exception as ex:
+    except Exception:
         raise HTTPException(
             status_code=403, detail=[{"msg": "OSM Authentication failed"}]
         )
@@ -43,11 +58,11 @@ def get_osm_auth_user(access_token):
     return user
 
 
-def login_required(access_token: str = Header(...)):
+def login_required(access_token: str = Depends(API_Access_Token)):
     return get_osm_auth_user(access_token)
 
 
-def get_optional_user(access_token: str = Header(default=None)) -> AuthUser:
+def get_optional_user(access_token: str | None = Depends(API_Access_Token)) -> AuthUser:
     if access_token:
         return get_osm_auth_user(access_token)
     else:
@@ -57,8 +72,8 @@ def get_optional_user(access_token: str = Header(default=None)) -> AuthUser:
 
 def admin_required(user: AuthUser = Depends(login_required)):
     db_user = get_user_from_db(user.id)
-    if not db_user["role"] is UserRole.ADMIN.value:
-        raise HTTPException(status_code=403, detail="User is not an admin")
+    if db_user["role"] is not UserRole.ADMIN.value:
+        raise HTTPException(status_code=403, detail=[{"msg": "User is not an admin"}])
     return user
 
 
@@ -70,5 +85,5 @@ def staff_required(user: AuthUser = Depends(login_required)):
         db_user["role"] is UserRole.STAFF.value
         or db_user["role"] is UserRole.ADMIN.value
     ):
-        raise HTTPException(status_code=403, detail="User is not a staff")
+        raise HTTPException(status_code=403, detail=[{"msg": "User is not a staff"}])
     return user

API/auth/routers.py

@@ -1,18 +1,33 @@
-import json
-
-from fastapi import APIRouter, Depends, Request
+from fastapi import APIRouter, Depends, Request, Path, Query
 from pydantic import BaseModel
 
 from src.app import Users
+from src.validation.models import ErrorMessage, common_responses
 
 from . import AuthUser, admin_required, login_required, osm_auth, staff_required
 
 router = APIRouter(prefix="/auth", tags=["Auth"])
 
 
-@router.get("/login/")
+@router.get(
+    "/login",
+    responses={
+        200: {
+            "description": "A Login URL",
+            "content": {
+                "application/json": {
+                    "example": {
+                        "login_url": "https://www.openstreetmap.org/oauth2/authorize/..."
+                    }
+                }
+            },
+        },
+        500: {"model": ErrorMessage},
+    },
+)
 def login_url(request: Request):
-    """Generate Login URL for authentication using OAuth2 Application registered with OpenStreetMap.
+    """
+    Generate Login URL for authentication using OAuth2 Application registered with OpenStreetMap.
     Click on the download url returned to get access_token.
 
     Parameters: None
@@ -25,7 +40,7 @@ def login_url(request: Request):
     return login_url
 
 
-@router.get("/callback/")
+@router.get("/callback", responses={500: {"model": ErrorMessage}})
 def callback(request: Request):
     """Performs token exchange between OpenStreetMap and Raw Data API
 
@@ -42,18 +57,27 @@ def callback(request: Request):
     return access_token
 
 
-@router.get("/me/", response_model=AuthUser)
+@router.get(
+    "/me",
+    response_model=AuthUser,
+    responses={**common_responses},
+    response_description="User Information",
+)
 def my_data(user_data: AuthUser = Depends(login_required)):
     """Read the access token and provide  user details from OSM user's API endpoint,
     also integrated with underpass .
 
     Parameters:None
 
-    Returns: user_data
+    Returns: user_data\n
             User Role :
                 ADMIN = 1
                 STAFF = 2
                 GUEST = 3
+
+    Raises:
+    - HTTPException 403: Due to authentication error(Wrong access token).
+    - HTTPException 500: Internal server error
     """
     return user_data
 
@@ -62,9 +86,19 @@ class User(BaseModel):
     osm_id: int
     role: int
 
+    class Config:
+        json_schema_extra = {"example": {"osm_id": 123, "role": 1}}
+
 
 # Create user
-@router.post("/users/", response_model=dict)
+@router.post(
+    "/users",
+    response_model=dict,
+    responses={
+        **common_responses,
+        "200": {"content": {"application/json": {"example": {"osm_id": 123}}}},
+    },
+)
 async def create_user(params: User, user_data: AuthUser = Depends(admin_required)):
     """
     Creates a new user and returns the user's information.
@@ -80,15 +114,26 @@ async def create_user(params: User, user_data: AuthUser = Depends(admin_required
     - Dict[str, Any]: A dictionary containing the osm_id of the newly created user.
 
     Raises:
-    - HTTPException: If the user creation fails.
+    - HTTPException 403: If the user creation fails due to insufficient permission.
+    - HTTPException 500: If the user creation fails due to internal server error.
     """
     auth = Users()
     return auth.create_user(params.osm_id, params.role)
 
 
 # Read user by osm_id
-@router.get("/users/{osm_id}", response_model=dict)
-async def read_user(osm_id: int, user_data: AuthUser = Depends(staff_required)):
+@router.get(
+    "/users/{osm_id}",
+    responses={
+        **common_responses,
+        "200": {"content": {"application/json": {"example": {"osm_id": 1, "role": 2}}}},
+        "404": {"model": ErrorMessage},
+    },
+)
+async def read_user(
+    osm_id: int = Path(description="The OSM ID of the User to Retrieve"),
+    user_data: AuthUser = Depends(staff_required),
+):
     """
     Retrieves user information based on the given osm_id.
     User Role :
@@ -103,17 +148,28 @@ async def read_user(osm_id: int, user_data: AuthUser = Depends(staff_required)):
     - Dict[str, Any]: A dictionary containing user information.
 
     Raises:
-    - HTTPException: If the user with the given osm_id is not found.
+    - HTTPException 403: If the user has insufficient permission.
+    - HTTPException 404: If the user with the given osm_id is not found.
+    - HTTPException 500: If it fails due to internal server error.
     """
     auth = Users()
 
     return auth.read_user(osm_id)
 
 
 # Update user by osm_id
-@router.put("/users/{osm_id}", response_model=dict)
+@router.put(
+    "/users/{osm_id}",
+    responses={
+        **common_responses,
+        "200": {"content": {"application/json": {"example": {"osm_id": 1, "role": 1}}}},
+        "404": {"model": ErrorMessage},
+    },
+)
 async def update_user(
-    osm_id: int, update_data: User, user_data: AuthUser = Depends(admin_required)
+    update_data: User,
+    user_data: AuthUser = Depends(admin_required),
+    osm_id: int = Path(description="The OSM ID of the User to Update"),
 ):
     """
     Updates user information based on the given osm_id.
@@ -129,15 +185,27 @@ async def update_user(
     - Dict[str, Any]: A dictionary containing the updated user information.
 
     Raises:
-    - HTTPException: If the user with the given osm_id is not found.
+    - HTTPException 403: If the user has insufficient permission.
+    - HTTPException 404: If the user with the given osm_id is not found.
+    - HTTPException 500: If it fails due to internal server error.
     """
     auth = Users()
     return auth.update_user(osm_id, update_data)
 
 
 # Delete user by osm_id
-@router.delete("/users/{osm_id}", response_model=dict)
-async def delete_user(osm_id: int, user_data: AuthUser = Depends(admin_required)):
+@router.delete(
+    "/users/{osm_id}",
+    responses={
+        **common_responses,
+        "200": {"content": {"application/json": {"example": {"osm_id": 1, "role": 1}}}},
+        "404": {"model": ErrorMessage},
+    },
+)
+async def delete_user(
+    user_data: AuthUser = Depends(admin_required),
+    osm_id: int = Path(description="The OSM ID of the User to Delete"),
+):
     """
     Deletes a user based on the given osm_id.
 
@@ -148,16 +216,29 @@ async def delete_user(osm_id: int, user_data: AuthUser = Depends(admin_required)
     - Dict[str, Any]: A dictionary containing the deleted user information.
 
     Raises:
-    - HTTPException: If the user with the given osm_id is not found.
+    - HTTPException 403: If the user has insufficient permission.
+    - HTTPException 404: If the user with the given osm_id is not found.
+    - HTTPException 500: If it fails due to internal server error.
     """
     auth = Users()
     return auth.delete_user(osm_id)
 
 
 # Get all users
-@router.get("/users/", response_model=list)
+@router.get(
+    "/users",
+    response_model=list,
+    responses={
+        **common_responses,
+        "200": {
+            "content": {"application/json": {"example": [{"osm_id": 1, "role": 2}]}}
+        },
+    },
+)
 async def read_users(
-    skip: int = 0, limit: int = 10, user_data: AuthUser = Depends(staff_required)
+    skip: int = Query(0, description="The Number of Users to Skip"),
+    limit: int = Query(10, description="The Maximum Number of Users to Retrieve"),
+    user_data: AuthUser = Depends(staff_required),
 ):
     """
     Retrieves a list of users with optional pagination.
@@ -168,6 +249,9 @@ async def read_users(
 
     Returns:
     - List[Dict[str, Any]]: A list of dictionaries containing user information.
+
+    - HTTPException 403: If it fails due to insufficient permission.
+    - HTTPException 500: If it fails due to internal server error.
     """
     auth = Users()
     return auth.read_users(skip, limit)

API/custom_exports.py

@@ -1,6 +1,4 @@
 # Standard library imports
-import json
-from typing import Dict
 
 # Third party imports
 import yaml
@@ -13,15 +11,34 @@
 from src.config import DEFAULT_QUEUE_NAME
 from src.config import LIMITER as limiter
 from src.config import RATE_LIMIT_PER_MIN
-from src.validation.models import CustomRequestsYaml, DynamicCategoriesModel
+from src.validation.models import (
+    CustomRequestsYaml,
+    DynamicCategoriesModel,
+    common_responses,
+)
 
 from .api_worker import process_custom_request
 from .auth import AuthUser, UserRole, staff_required
 
 router = APIRouter(prefix="/custom", tags=["Custom Exports"])
 
 
-@router.post("/snapshot/")
+@router.post(
+    "/snapshot",
+    responses={
+        **common_responses,
+        "200": {
+            "content": {
+                "application/json": {
+                    "example": {
+                        "task_id": "3fded368-456f-4ef4-a1b8-c099a7f77ca4",
+                        "track_link": "/tasks/status/3fded368-456f-4ef4-a1b8-c099a7f77ca4/",
+                    }
+                }
+            }
+        },
+    },
+)
 @limiter.limit(f"{RATE_LIMIT_PER_MIN}/minute")
 @version(1)
 async def process_custom_requests(
@@ -827,7 +844,7 @@ async def process_custom_requests(
 
 
 @router.post(
-    "/snapshot/yaml/",
+    "/snapshot/yaml",
     openapi_extra={
         "requestBody": {
             "content": {
@@ -847,11 +864,13 @@ async def process_custom_requests_yaml(
     try:
         data = yaml.safe_load(raw_body)
     except yaml.YAMLError:
-        raise HTTPException(status_code=422, detail="Invalid YAML")
+        raise HTTPException(status_code=422, detail=[{"msg": "Invalid YAML"}])
     try:
         validated_data = DynamicCategoriesModel.model_validate(data)
     except ValidationError as e:
-        raise HTTPException(status_code=422, detail=e.errors(include_url=False))
+        raise HTTPException(
+            status_code=422, detail=[{"msg": e.errors(include_url=False)}]
+        )
 
     queue_name = validated_data.queue
     if validated_data.queue != DEFAULT_QUEUE_NAME and user.role != UserRole.ADMIN.value: