Repurpose the project to use database for secret storage

.gitignore

@@ -163,3 +163,4 @@ cython_debug/
 
 *.env
 *.json
+*.db

vaultapi/__init__.py

@@ -3,9 +3,7 @@
 import click
 from cryptography.fernet import Fernet
 
-from vaultapi.main import start  # noqa: F401
-
-version = "0.0.0-a"
+from .main import start, version
 
 
 @click.command()
@@ -46,7 +44,7 @@ def commandline(*args, **kwargs) -> None:
         for k, v in options.items()
     )
     if kwargs.get("version"):
-        click.echo(f"vaultapi {version}")
+        click.echo(f"vaultapi {version.__version__}")
         sys.exit(0)
     if kwargs.get("help"):
         click.echo(

vaultapi/auth.py

@@ -6,7 +6,7 @@
 from fastapi import Request
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 
-from vaultapi import exceptions, models
+from . import exceptions, models
 
 LOGGER = logging.getLogger("uvicorn.default")
 EPOCH = lambda: int(time.time())  # noqa: E731

vaultapi/database.py

@@ -0,0 +1,51 @@
+from . import models
+
+
+def get_secret(key: str, table_name: str) -> str | None:
+    """Function to retrieve secret from database.
+
+    Args:
+        key: Name of the secret to retrieve.
+        table_name: Name of the table where the secret is stored.
+
+    Returns:
+        str:
+        Returns the secret value.
+    """
+    with models.database.connection:
+        cursor = models.database.connection.cursor()
+        state = cursor.execute(
+            f'SELECT value FROM "{table_name}" WHERE key=(?)', (key,)
+        ).fetchone()
+    if state and state[0]:
+        return state[0]
+
+
+def put_secret(key: str, value: str, table_name: str) -> None:
+    """Function to add secret to the database.
+
+    Args:
+        key: Name of the secret to be stored.
+        value: Value of the secret to be stored
+        table_name: Name of the table where the secret is stored.
+    """
+    with models.database.connection:
+        cursor = models.database.connection.cursor()
+        cursor.execute(
+            f'INSERT INTO "{table_name}" (key, value) VALUES (?,?)',
+            (key, value),
+        )
+        models.database.connection.commit()
+
+
+def remove_secret(key: str, table_name: str) -> None:
+    """Function to remove a secret from the database.
+
+    Args:
+        key: Name of the secret to be removed.
+        table_name: Name of the table where the secret is stored.
+    """
+    with models.database.connection:
+        cursor = models.database.connection.cursor()
+        cursor.execute(f'DELETE FROM "{table_name}" WHERE key=(?)', (key,))
+        models.database.connection.commit()

vaultapi/main.py

@@ -1,13 +1,17 @@
 import logging
+import pathlib
 
 import uvicorn
-from cryptography.fernet import Fernet
 from fastapi import FastAPI
 
-import vaultapi
-from vaultapi import models, routers, squire
+from . import models, routers, squire, version
 
 LOGGER = logging.getLogger("uvicorn.default")
+VaultAPI = FastAPI(
+    title="VaultAPI",
+    description="Lightweight service to serve secrets and environment variables",
+    version=version.__version__,
+)
 
 
 def start(**kwargs) -> None:
@@ -25,18 +29,16 @@ def start(**kwargs) -> None:
         log_config: Logging configuration as a dict or a FilePath. Supports .yaml/.yml, .json or .ini formats.
     """
     models.env = squire.load_env(**kwargs)
-    models.session.fernet = Fernet(models.env.secret)
-    app = FastAPI(
-        routes=routers.get_all_routes(),
-        title="vaultapi",
-        description="Lightweight service to serve secrets and environment variables",
-        version=vaultapi.version,
-    )
+    # models.session.fernet = Fernet(models.env.secret)
+    models.database = models.Database(models.env.database)
+    models.database.create_table("default", ["key", "value"])
+    module_name = pathlib.Path(__file__)
+    VaultAPI.routes.extend(routers.get_all_routes())
     kwargs = dict(
         host=models.env.host,
         port=models.env.port,
         workers=models.env.workers,
-        app=app,
+        app=f"{module_name.parent.stem}.{module_name.stem}:{VaultAPI.title}",
     )
     if models.env.log_config:
         kwargs["log_config"] = models.env.log_config

vaultapi/models.py

@@ -1,12 +1,49 @@
 import pathlib
 import socket
-from typing import Any, Dict, List, Set
+import sqlite3
+from typing import Any, Dict, List, Set, Tuple
 
 from cryptography.fernet import Fernet
-from pydantic import BaseModel, DirectoryPath, FilePath, PositiveInt
+from pydantic import BaseModel, Field, FilePath, PositiveInt
 from pydantic_settings import BaseSettings
 
 
+class Database:
+    """Creates a connection and instantiates the cursor.
+
+    >>> Database
+
+    Args:
+        filepath: Name of the database file.
+        timeout: Timeout for the connection to database.
+    """
+
+    def __init__(self, filepath: FilePath | str, timeout: int = 10):
+        """Instantiates the class ``Database`` to create a connection and a cursor."""
+        if not filepath.endswith(".db"):
+            filepath = filepath + ".db"
+        self.connection = sqlite3.connect(
+            database=filepath, check_same_thread=False, timeout=timeout
+        )
+
+    def create_table(self, table_name: str, columns: List[str] | Tuple[str]) -> None:
+        """Creates the table with the required columns.
+
+        Args:
+            table_name: Name of the table that has to be created.
+            columns: List of columns that has to be created.
+        """
+        with self.connection:
+            cursor = self.connection.cursor()
+            # Use f-string or %s as table names cannot be parametrized
+            cursor.execute(
+                f"CREATE TABLE IF NOT EXISTS {table_name!r} ({', '.join(columns)})"
+            )
+
+
+database: Database = Database  # noqa: PyTypeChecker
+
+
 class RateLimit(BaseModel):
     """Object to store the rate limit settings.
 
@@ -44,14 +81,14 @@ class EnvConfig(BaseSettings):
     """
 
     apikey: str
-    secret: str
+    # secret: str
+    database: str = Field("secrets.db", pattern=".*.db$")
     host: str = socket.gethostbyname("localhost") or "0.0.0.0"
     port: PositiveInt = 8080
     workers: PositiveInt = 1
     log_config: FilePath | Dict[str, Any] | None = None
     allowed_origins: List[str] = []
     rate_limit: RateLimit | List[RateLimit] = []
-    secrets_path: DirectoryPath = "secrets"
 
     @classmethod
     def from_env_file(cls, env_file: pathlib.Path) -> "EnvConfig":
@@ -74,5 +111,6 @@ class Config:
         arbitrary_types_allowed = True
 
 
-env = EnvConfig
+# noinspection PyTypeChecker
+env: EnvConfig = EnvConfig
 session = Session()

vaultapi/payload.py

@@ -0,0 +1,24 @@
+from pydantic import BaseModel, FilePath, DirectoryPath
+
+
+class DeleteSecret(BaseModel):
+    """Payload for delete-secret API call.
+
+    >>> DeleteSecret
+
+    """
+
+    key: str
+    table_name: str = "default"
+
+
+class PutSecret(BaseModel):
+    """Payload for put-secret API call.
+
+    >>> DeleteSecret
+
+    """
+
+    key: str
+    value: str
+    table_name: str = "default"

vaultapi/rate_limit.py

@@ -4,7 +4,7 @@
 
 from fastapi import HTTPException, Request
 
-from vaultapi import models
+from . import models
 
 
 class RateLimiter: