Renamed project to pssecret-server

pssecret_server/__init__.py

@@ -0,0 +1,5 @@
+__all__ = [
+    "cli",
+]
+
+from .cli import cli

pssecret_server/cli.py

@@ -0,0 +1,6 @@
+import click
+
+
+@click.command()
+def cli():
+    print("Hello, world")

pssecret_server/main.py

@@ -0,0 +1,51 @@
+from typing import Annotated
+
+from fastapi import Depends, FastAPI
+from fastapi.exceptions import HTTPException
+from redis.asyncio import Redis
+
+from pssecret_server.models import Secret, SecretSaveResult
+from pssecret_server.redis_db import get_redis
+from pssecret_server.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,
+    }

pssecret_server/models.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel, Field
+
+
+class Secret(BaseModel):
+    data: str = Field(title="Secret", description="Some secret data")
+
+
+class SecretSaveResult(BaseModel):
+    key: str = Field(
+        title="Retrieval key",
+        description="Key that should be used for retrieval of submitted secret",
+    )

pssecret_server/redis_db.py

@@ -0,0 +1,11 @@
+# noinspection PyUnresolvedReferences,PyProtectedMember
+from typing import Annotated
+
+from fastapi import Depends
+from redis import asyncio as aioredis
+
+from pssecret_server.settings import Settings, get_settings
+
+
+def get_redis(settings: Annotated[Settings, Depends(get_settings)]) -> aioredis.Redis:
+    return aioredis.from_url(str(settings.redis_url))

pssecret_server/settings.py

@@ -0,0 +1,10 @@
+from pydantic import RedisDsn
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    redis_url: RedisDsn = RedisDsn("redis://localhost")
+
+
+def get_settings() -> Settings:
+    return Settings()

pssecret_server/utils.py

@@ -0,0 +1,22 @@
+from uuid import uuid4
+
+from redis.asyncio import Redis
+
+from pssecret_server.models import Secret
+
+
+async def get_new_key(redis: Redis) -> str:
+    """Returns free Redis key"""
+    while True:
+        new_key = str(uuid4())
+
+        if not await redis.exists(new_key):
+            return new_key
+
+
+async def save_secret(data: Secret, redis: Redis) -> str:
+    """Save passed data, returns retrieval key"""
+    new_key = await get_new_key(redis)
+    await redis.setex(new_key, 60 * 60 * 24, data.data)
+
+    return new_key