pssecret-server/pssecret/main.py
Ivan Golikov 8266c95cd9 Integration tests (#3)
Reviewed-on: root/pssecret#3
Co-authored-by: Ivan Golikov <root@ivnglkv.me>
Co-committed-by: Ivan Golikov <root@ivnglkv.me>
2025-01-01 18:18:38 +00:00

51 lines
1.4 KiB
Python

from typing import Annotated
from fastapi import Depends, FastAPI
from fastapi.exceptions import HTTPException
from redis.asyncio import Redis
from pssecret.models import Secret, SecretSaveResult
from pssecret.redis_db import get_redis
from pssecret.utils import save_secret
app = FastAPI()
RedisDep = Annotated[Redis, Depends(get_redis)]
@app.post(
"/secret",
summary="Store secret",
description=(
"Submit secret, it is saved on the server, get retrieval key in response. "
"Use that key to retrieve your data. Key could be used only once, "
"so use it wisely"
),
response_model=SecretSaveResult,
)
async def set_secret(data: Secret, redis: RedisDep) -> dict[str, str]:
return {
"key": await save_secret(data, redis),
}
@app.get(
"/secret/{secret_key}",
summary="Retrieve secret",
description=(
"Returns previously saved data if it is still on the server. "
"Could be the other way around in two cases: "
"either it has already been retrieved, either storage timeout has expired"
),
response_model=Secret,
responses={404: {"description": "The item was not found"}},
)
async def get_secret(secret_key: str, redis: RedisDep) -> dict[str, bytes]:
data: bytes | None = await redis.getdel(secret_key)
if data is None:
raise HTTPException(404)
return {
"data": data,
}