Migrated to poetry

pssecret/__init__.py

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

pssecret/cli.py

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

pssecret/main.py

@@ -0,0 +1,36 @@
+from fastapi import FastAPI
+from fastapi.exceptions import HTTPException
+
+from pssecret.models import Secret, SecretSaveResult
+from pssecret.redis_db import redis
+from pssecret.utils import get_new_key
+
+app = FastAPI()
+
+
+@app.post("/secret", response_model=SecretSaveResult)
+async def set_secret(data: Secret):
+    new_key = await get_new_key()
+    await redis.setex(new_key, 60 * 60 * 24, data.data)
+
+    return {
+        "status": "saved",
+        "retrieval_url": f"/secret/{new_key}",
+    }
+
+
+@app.get(
+    "/secret/{secret_key}",
+    response_model=Secret,
+    responses={404: {"description": "The item was not found"}},
+)
+async def get_secret(secret_key):
+    data = await redis.get(secret_key)
+
+    if data is None:
+        raise HTTPException(404)
+
+    await redis.delete(secret_key)
+    return {
+        "data": data,
+    }

pssecret/models.py

@@ -0,0 +1,10 @@
+from pydantic import BaseModel
+
+
+class Secret(BaseModel):
+    data: str
+
+
+class SecretSaveResult(BaseModel):
+    status: str
+    retrieval_url: str

pssecret/redis_db.py

@@ -0,0 +1,6 @@
+# noinspection PyUnresolvedReferences,PyProtectedMember
+from redis import asyncio as aioredis
+
+from pssecret.settings import settings
+
+redis = aioredis.from_url(settings.redis.url)

pssecret/settings.py

@@ -0,0 +1,26 @@
+import os
+import tomllib
+
+
+class Settings:
+    def __init__(self, data: dict = None):
+        if data:
+            self._data = data
+        else:
+            with open(
+                os.getenv("PSSECRET_CONF_FILE", "/etc/pssecret/pssecret.toml"), "rb"
+            ) as f:
+                self._data = tomllib.load(f)
+
+    def __getattr__(self, item):
+        try:
+            value = self._data[item]
+        except KeyError:
+            raise AttributeError
+        if isinstance(value, dict):
+            return Settings(data=value)
+        else:
+            return value
+
+
+settings = Settings()

pssecret/utils.py

@@ -0,0 +1,11 @@
+from uuid import uuid4
+
+from pssecret.redis_db import redis
+
+
+async def get_new_key() -> str:
+    while True:
+        new_key = str(uuid4())
+
+        if not await redis.exists(new_key):
+            return new_key