85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
from fastapi import FastAPI, Request, Response, Depends
|
|
from fastapi_throttle import RateLimiter
|
|
from fastapi.responses import JSONResponse
|
|
from utils.meme_util import MemeUtil
|
|
|
|
|
|
class Meme(FastAPI):
|
|
"""
|
|
Meme Endpoints
|
|
"""
|
|
|
|
def __init__(self, app: FastAPI, my_util, constants) -> None:
|
|
"""Initialize Meme endpoints."""
|
|
self.app: FastAPI = app
|
|
self.util = my_util
|
|
self.meme_util = MemeUtil(constants)
|
|
self.constants = constants
|
|
self.endpoints: dict = {
|
|
"memes/get_meme/{id:path}": self.get_meme_by_id,
|
|
"memes/random": self.random_meme,
|
|
"memes/list_memes": self.list_memes,
|
|
}
|
|
|
|
for endpoint, handler in self.endpoints.items():
|
|
dependencies = None
|
|
if endpoint == "memes/list_memes":
|
|
dependencies = [
|
|
Depends(RateLimiter(times=10, seconds=2))
|
|
] # Do not rate limit image retrievals (cached)
|
|
app.add_api_route(
|
|
f"/{endpoint}",
|
|
handler,
|
|
methods=["GET"],
|
|
include_in_schema=False,
|
|
dependencies=dependencies,
|
|
)
|
|
|
|
async def get_meme_by_id(self, id: int, request: Request) -> Response:
|
|
"""
|
|
Get meme image by ID.
|
|
|
|
Parameters:
|
|
- **id** (int): Meme ID.
|
|
- **request** (Request): The request object.
|
|
|
|
Returns:
|
|
- **Response**: Image response or 404.
|
|
"""
|
|
meme_image = await self.meme_util.get_meme_by_id(id)
|
|
if not meme_image:
|
|
return Response(status_code=404, content="Not found")
|
|
return Response(content=meme_image, media_type="image/png")
|
|
|
|
async def random_meme(self, request: Request) -> Response:
|
|
"""
|
|
Get a random meme image.
|
|
|
|
Parameters:
|
|
- **request** (Request): The request object.
|
|
|
|
Returns:
|
|
- **Response**: Image response or 404.
|
|
"""
|
|
meme_image = await self.meme_util.get_random_meme()
|
|
if not meme_image:
|
|
return Response(status_code=404, content="Not found")
|
|
return Response(content=meme_image, media_type="image/png")
|
|
|
|
async def list_memes(self, page: int, request: Request) -> Response:
|
|
"""
|
|
List memes with pagination.
|
|
|
|
Parameters:
|
|
- **page** (int): Page number.
|
|
- **request** (Request): The request object.
|
|
|
|
Returns:
|
|
- **JSONResponse**: List of memes with paging info.
|
|
"""
|
|
meme_list = await self.meme_util.list_memes(page)
|
|
page_count = await self.meme_util.get_page_count()
|
|
return JSONResponse(
|
|
content={"paging": {"current": page, "of": page_count}, "memes": meme_list}
|
|
)
|