FastAPI RESTPlus

---

1. Creating a Simple API with CRUD Operations

from fastapi import FastAPI
from fastapi.restful import Resource, Api

app = FastAPI()
api = Api(app)

class Item(Resource):
    def get(self, item_id):
        return {"item_id": item_id}

    def post(self, item_id):
        return {"item_id": item_id}

api.add_resource(Item, "/items/{item_id}")

2. Using Request Validation

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from pydantic import BaseModel

class ItemModel(BaseModel):
    name: str

class Item(Resource):
    def post(self, item: ItemModel):
        return item.dict()

api = Api(app)
api.add_resource(Item, "/items")

3. Handling Authentication with JWT

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.security import HTTPBearer

oauth2_scheme = HTTPBearer()

class Item(Resource):
    @oauth2_scheme.required
    def get(self):
        return {"message": "Protected resource"}

api = Api(app)
api.add_resource(Item, "/items")

4. Using CORS

from fastapi import FastAPI
from fastapi.restful import Resource, Api

app = FastAPI()
api = Api(app)

class Item(Resource):
    def options(self):
        return {"Access-Control-Allow-Origin": "*"}

api.add_resource(Item, "/items")

5. Handling File Uploads

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.responses import FileResponse

class Upload(Resource):
    def post(self):
        file = request.files.get("file")
        file.save("uploaded_file.txt")
        return FileResponse("uploaded_file.txt")

api = Api(app)
api.add_resource(Upload, "/upload")

6. Building a RESTful API with Pagination

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from typing import List

class Item(Resource):
    def get(self, page_number: int, page_size: int):
        items = get_items(page_number, page_size)
        return {"items": items, "total_count": get_total_count()}

api = Api(app)
api.add_resource(Item, "/items")

7. Implementing a Filtering System

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from typing import List

class Item(Resource):
    def get(self, name: str, category: str):
        items = get_items_by_filter(name, category)
        return {"items": items}

api = Api(app)
api.add_resource(Item, "/items")

8. Handling Dynamically Generated Resources

from fastapi import FastAPI
from fastapi.restful import Resource, Api, abort

class DynamicResource(Resource):
    def get(self, resource_name):
        if resource_name not in resources:
            abort(404)
        return resources[resource_name]

api = Api(app)
api.add_resource(DynamicResource, "/resources/{resource_name}")

9. Creating a Versioned API

from fastapi import FastAPI
from fastapi.restful import Resource, Api

app = FastAPI()
api_v1 = Api(app, prefix="/api/v1")
api_v2 = Api(app, prefix="/api/v2")

class Item(Resource):
    def get(self):
        return {"message": "Hello from API version 1"}

class ItemV2(Resource):
    def get(self):
        return {"message": "Hello from API version 2"}

api_v1.add_resource(Item, "/items")
api_v2.add_resource(ItemV2, "/items")

10. Building a RESTful API with OpenAPI 3 Documentation

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.responses import JSONResponse

openapi_schema = {
    "info": {
        "title": "My RESTful API",
        "description": "This is an OpenAPI 3 schema for my RESTful API.",
    },
    "paths": {
        "/items": {
            "get": {"responses": {"200": {"description": "OK"}}},
            "post": {"responses": {"201": {"description": "Created"}}},
        },
    },
}

app = FastAPI(openapi_url="/openapi.json", openapi_schema=openapi_schema)
api = Api(app)

class Item(Resource):
    def get(self):
        return JSONResponse(content={"message": "Hello from RESTful API"}, status_code=200)

api.add_resource(Item, "/items")

11. Using a Custom Request Parser

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

class CustomRequestParser:
    def parse_args(self):
        return jsonable_encoder(request.json())

class Item(Resource):
    def post(self):
        parser = CustomRequestParser()
        data = parser.parse_args()
        return {"message": data}

api = Api(app)
api.add_resource(Item, "/items")

12. Handling Errors with Exception Handling

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.responses import JSONResponse

class Item(Resource):
    def get(self, item_id):
        try:
            item = get_item(item_id)
            return JSONResponse(content={"item": item}, status_code=200)
        except ItemNotFoundError:
            return JSONResponse(content={"error": "Item not found"}, status_code=404)

api = Api(app)
api.add_resource(Item, "/items/{item_id}")

13. Building a RESTful API with Swagger UI

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.responses import RedirectResponse

app = FastAPI()
api = Api(app)

class Redirect(Resource):
    def get(self):
        return RedirectResponse("/swagger-ui")

api.add_resource(Redirect, "/docs")

14. Using JSON Schemas for Request and Response Validation

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from pydantic import BaseModel

class ItemModel(BaseModel):
    name: str

class Item(Resource):
    def post(self, item: ItemModel):
        return item.dict()

api = Api(app)
api.add_resource(Item, "/items")

15. Implementing a Request Interceptor

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.requests import Request

class RequestInterceptor:
    def __call__(self, request: Request):
        print(f"Request received: {request.url}")

class Item(Resource):
    def get(self):
        return {"message": "Hello from RESTful API"}

api = Api(app, request_interceptor=RequestInterceptor())
api.add_resource(Item, "/items")

16. Handling Rate Limiting

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.responses import HTTP_429_TOO_MANY_REQUESTS

class RateLimitMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if request.url.endswith("/api/endpoint"):
            # Check rate limit for endpoint
            if rate_limit_exceeded:
                return HTTP_429_TOO_MANY_REQUESTS

        # Pass request to downstream
        return await self.app(scope, receive, send)

app = FastAPI(middleware_class=RateLimitMiddleware)
api = Api(app)

class Item(Resource):
    def get(self):
        return {"message": "Hello from RESTful API"}

api.add_resource(Item, "/api/endpoint")

17. Using Dependecies for Resource Initialization

from fastapi import FastAPI, Depends
from fastapi.restful import Resource, Api

class CustomDependency:
    def __init__(self, value):
        self.value = value

class Item(Resource):
    def __init__(self, dependency: CustomDependency = Depends()):
        self.dependency = dependency

api = Api(app)
api.add_resource(Item, "/items")

18. Building a RESTful API with Async Support

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.responses import AsyncJSONResponse

class Item(Resource):
    async def get(self):
        items = await get_items()
        return AsyncJSONResponse(content={"items": items}, status_code=200)

api = Api(app)
api.add_resource(Item, "/items")

19. Implementing Authentication with OAuth2

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.security.oauth2 import OAuth2PasswordRequestForm

class Token(Resource):
    def post(self, form_data: OAuth2PasswordRequestForm = Depends()):
        return {"access_token": "fake-token"}

api = Api(app)
api.add_resource(Token, "/token")

20. Handling File Uploads with Multipart Form Data

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.responses import FileResponse

class Upload(Resource):
    def post(self):
        file = request.files.get("file")
        file.save("uploaded_file.txt")
        return FileResponse("uploaded_file.txt")

api = Api(app)
api.add_resource(Upload, "/upload")

21. Building a RESTful API with CORS Support

from fastapi import FastAPI
from fastapi.restful import Resource, Api

class CORSResource(Resource):
    def options(self, *args, **kwargs):
        return make_response("", 200)

api = Api(app)
api.add_resource(CORSResource, "/items")

22. Using JWT for Authentication and Authorization

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.security import HTTPBearer

oauth2_scheme = HTTPBearer()

class Item(Resource):
    @oauth2_scheme.required
    def get(self):
        return {"message": "Protected resource"}

api = Api(app)
api.add_resource(Item, "/items")

23. Building a RESTful API with Pagination and Filtering

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from typing import List

class Item(Resource):
    def get(self, page_number: int, page_size: int):
        items = get_items(page_number, page_size)
        return {"items": items, "total_count": get_total_count()}

    def post(self, name: str, category: str):
        items = get_items_by_filter(name, category)
        return {"items": items}

api = Api(app)
api.add_resource(Item, "/items")

24. Implementing Versioning in a RESTful API

from fastapi import FastAPI
from fastapi.restful import Resource, Api

app = FastAPI()
api_v1 = Api(app, prefix="/api/v1")
api_v2 = Api(app, prefix="/api/v2")

class ItemV1(Resource):
    def get(self):
        return {"message": "Hello from API version 1"}

class ItemV2(Resource):
    def get(self):
        return {"message": "Hello from API version 2"}

api_v1.add_resource(ItemV1, "/items")
api_v2.add_resource(ItemV2, "/items")

25. Creating a RESTful API with Swagger Documentation

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.responses import JSONResponse

openapi_schema = {
    "info": {
        "title": "My RESTful API",
        "description": "This is an OpenAPI 3 schema for my RESTful API.",
    },
    "paths": {
        "/items": {
            "get": {"responses": {"200": {"description": "OK"}}},
            "post": {"responses": {"201": {"description": "Created"}}},
        },
    },
}

app = FastAPI(openapi_url="/openapi.json", openapi_schema=openapi_schema)
api = Api(app)

class Item(Resource):
    def get(self):
        return JSONResponse(content={"message": "Hello from RESTful API"}, status_code=200)

api.add_resource(Item, "/items")

26. Building a RESTful API with Custom Request and Response Parsers

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from pydantic import BaseModel

class CustomRequestParser:
    def parse_args(self):
        return {"name": request.args.get("name")}

class CustomResponseParser:
    def render(self, data):
        return JSONResponse(content={"message": data}, status_code=200)

class Item(Resource):
    def get(self, parser: CustomRequestParser = Depends()):
        args = parser.parse_args()
        return {"message": f"Hello {args['name']}"}

api = Api(app, request_parsers=[CustomRequestParser()], response_parsers=[CustomResponseParser()])
api.add_resource(Item, "/items")

27. Implementing Error Handling in a RESTful API

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.responses import JSONResponse

class Item(Resource):
    def get(self, item_id: int):
        try:
            item = get_item(item_id)
            return JSONResponse(content={"item": item}, status_code=200)
        except ItemNotFoundError:
            return JSONResponse(content={"error": "Item not found"}, status_code=404)

api = Api(app)
api.add_resource(Item, "/items/{item_id}")

28. Building a RESTful API with Dependency Injection

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.dependencies import Depends

class ItemDependency:
    def __init__(self, value):
        self.value = value

class Item(Resource):
    def __init__(self, dependency: ItemDependency = Depends()):
        self.dependency = dependency

api = Api(app)
api.add_resource(Item, "/items")

29. Creating a RESTful API with Rate Limiting

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.responses import HTTP_429_TOO_MANY_REQUESTS

limiter = RateLimit()

class Item(Resource):
    @limiter.limit("10/minute")
    def get(self):
        return {"message": "Hello from RESTful API"}

api = Api(app)
api.add_resource(Item, "/items")

30. Building a RESTful API with Background Tasks

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.background import BackgroundTasks

class Item(Resource):
    def __init__(self, background_tasks: BackgroundTasks):
        self.background_tasks = background_tasks

    def get(self, item_id: int):
        self.background_tasks.add_task(process_item, item_id)
        return {"message": "Item processed in background"}

api = Api(app)
api.add_resource(Item, "/items/{item_id}")

31. Implementing WebSockets in a RESTful API

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.websockets import WebSocket

class Item(Resource):
    def websocket(self, websocket: WebSocket):
        while True:
            message = await websocket.receive()
            await websocket.send(json.dumps({"message": message}))

api = Api(app)
api.add_resource(Item, "/items/websocket")

32. Building a RESTful API with Multiple Versions

from fastapi import FastAPI
from fastapi.restful import Resource, Api

app = FastAPI()
api_v1 = Api(app, prefix="/api/v1")
api_v2 = Api(app, prefix="/api/v2")

class ItemV1(Resource):
    def get(self):
        return {"message": "Hello from API version 1"}

class ItemV2(Resource):
    def get(self):
        return {"message": "Hello from API version 2"}

api_v1.add_resource(ItemV1, "/items")
api_v2.add_resource(ItemV2, "/items")

33. Creating a RESTful API with Blueprint Structure

from fastapi import FastAPI
from fastapi.restful import Api

app = FastAPI()
api = Api(app, prefix="/api")

class Item(Resource):
    def get(self):
        return {"message": "Hello from RESTful API"}

api.add_resource(Item, "/items")

34. Building a RESTful API with Event Handlers

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.events import Request, Response

class Item(Resource):
    def __init__(self):
        super().__init__()
        self.before_request_count = 0
        self.after_request_count = 0

    @fastapi.on_event("request")
    def before_request(self, request: Request):
        self.before_request_count += 1

    @fastapi.on_event("response")
    def after_request(self, request: Request, response: Response):
        self.after_request_count += 1

api = Api(app)
api.add_resource(Item, "/items")

35. Implementing Authentication and Authorization with JWT

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from fastapi.security import HTTPBearer

oauth2_scheme = HTTPBearer()

class Item(Resource):
    @oauth2_scheme.required
    def get(self):
        return {"message": "Protected resource"}

api = Api(app)
api.add_resource(Item, "/items")

36. Building a RESTful API with FastAPI and Tortoise ORM

from fastapi import FastAPI
from fastapi.restful import Resource, Api
from tortoise.contrib.fastapi import register_tortoise

app = FastAPI()
api = Api(app)

class Item(Resource):
    def get(self, item_id: int):
        item = await ItemModel.get(id=item_id)
        return {"name": item.name}

api.add_resource(Item, "/items/{item_id}")

register_tortoise(
    app,
    db_url="sqlite://db.sqlite3",
    modules={"models": ["models"]},
    generate_schemas=True,
    add_exception_handlers=True,
)