38 lines
1.4 KiB
Python
38 lines
1.4 KiB
Python
|
from fastapi import FastAPI, HTTPException, Depends
|
||
|
|
from pydantic import BaseModel
|
||
|
|
from aioredis import Redis
|
||
|
|
from aioredis.client import Redis as RedisClient
|
||
|
|
import aioredis
|
||
|
|
|
||
|
|
app = FastAPI()
|
||
|
|
|
||
|
|
REDIS_URL = "redis://localhost:6379"
|
||
|
|
REDIS_KEY = "stored_string"
|
||
|
|
|
||
|
|
async def get_redis() -> RedisClient:
|
||
|
|
return await aioredis.from_url(REDIS_URL, encoding="utf-8", decode_responses=True)
|
||
|
|
|
||
|
|
class StringRequest(BaseModel):
|
||
|
|
text: str
|
||
|
|
|
||
|
|
@app.post("/string", status_code=201)
|
||
|
|
async def create_string(request: StringRequest, redis: Redis = Depends(get_redis)):
|
||
|
|
if await redis.exists(REDIS_KEY):
|
||
|
|
raise HTTPException(status_code=400, detail="String already exists. Use PATCH to update.")
|
||
|
|
await redis.set(REDIS_KEY, request.text)
|
||
|
|
return {"message": "String created successfully", "text": request.text}
|
||
|
|
|
||
|
|
@app.patch("/string")
|
||
|
|
async def update_string(request: StringRequest, redis: Redis = Depends(get_redis)):
|
||
|
|
if not await redis.exists(REDIS_KEY):
|
||
|
|
raise HTTPException(status_code=404, detail="No string found to update. Use POST to create one.")
|
||
|
|
await redis.set(REDIS_KEY, request.text)
|
||
|
|
return {"message": "String updated successfully", "text": request.text}
|
||
|
|
|
||
|
|
@app.get("/string")
|
||
|
|
async def get_string(redis: Redis = Depends(get_redis)):
|
||
|
|
stored_string = await redis.get(REDIS_KEY)
|
||
|
|
if stored_string is None:
|
||
|
|
raise HTTPException(status_code=404, detail="No string found.")
|
||
|
|
return {"text": stored_string}
|