URL Expiration

pyproject.toml

@@ -16,13 +16,14 @@ dependencies = [
     "SQLAlchemy>=2.0.25",
     "psycopg2-binary>=2.9.9",
     "validators>=0.22.0",
+    "aiosqlite>=0.19.0",
 ]
 requires-python = ">=3.11.6"
 readme = "README.md"
 license = {text = "MIT"}
 
 [tool.pdm.scripts]
-start = "uvicorn src.link_liberate.main:app --host 0.0.0.0 --port 8080 --workers 1 --reload"
+start = "uvicorn src.link_liberate.main:app --host 0.0.0.0 --port 8080 --workers 4 --reload"
 dev = "uvicorn src.link_liberate.main:app --host 0.0.0.0 --port 8080 --reload"
 test = "pytest"
 mypy = "mypy src/link_liberate"

requirements.txt

@@ -62,3 +62,5 @@ wrapt==1.16.0
 
 SQLAlchemy~=2.0.25
 validators~=0.22.0
+
+aiosqlite~=0.19.0

src/link_liberate/database.py

@@ -2,6 +2,9 @@
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import sessionmaker
 
+import aiosqlite
+import asyncio
+
 SQLALCHEMY_DATABASE_URL = "sqlite:///./liberate.db"
 
 engine = create_engine(
@@ -19,3 +22,16 @@ def get_db():
         yield db
     finally:
         db.close()
+
+
+async def create_async_session():
+    async with aiosqlite.connect(SQLALCHEMY_DATABASE_URL) as db:
+        yield db
+
+
+async def expire_uuid(uuid):
+    print("Expiring uuid:", uuid)
+    await asyncio.sleep(60)  # Default expiration time is 60 mins
+    async with create_async_session() as db:
+        await db.execute("DELETE FROM liberatedlinks WHERE uuid = ?", (uuid,))
+        await db.commit()

src/link_liberate/main.py

@@ -16,12 +16,12 @@
 
 from slowapi.util import get_remote_address
 from sqlalchemy.orm import Session
-
+from asyncio import create_task
 from starlette.templating import _TemplateResponse
 
 from .utils import generate_uuid, make_proper_url, check_link
 from .models import Base, LiberatedLink
-from .database import engine, get_db
+from .database import engine, get_db, expire_uuid
 
 limiter = Limiter(key_func=get_remote_address)
 app: FastAPI = FastAPI(title="link-liberate")
@@ -77,6 +77,7 @@ async def web_post(
         db.add(new_liberated_link)
         db.commit()
         db.refresh(new_liberated_link)
+        await create_task(expire_uuid(uuid))
         context = {"link": link, "short": f"{BASE_URL}/{uuid}"}
     except Exception as e:
         raise HTTPException(

src/link_liberate/templates/base.html

@@ -11,7 +11,7 @@
             margin: 0;
             padding: 0;
         }
-        
+
         .container {
             max-width: 600px;
             align-content: center;
@@ -21,20 +21,20 @@
             border-radius: 8px;
             box-shadow: 0 0 10px rgba(0, 0, 0, 0.1);
         }
-        
+
         a {
           text-decoration: none !important;
         }
-        
+
         h1 {
             text-align: center;
             color: #333;
         }
-        
+
         .form-container {
             text-align: center;
         }
-        
+
         .link-input {
             width: 100%;
             padding: 10px;
@@ -43,7 +43,7 @@
             border: 1px solid #ccc;
             border-radius: 4px;
         }
-        
+
         .submit-button {
             background-color: #4CAF50;
             color: #fff;
@@ -53,11 +53,11 @@
             cursor: pointer;
             font-size: 16px;
         }
-        
+
         .submit-button:hover {
             background-color: #45a049;
         }
-        
+
     </style>
     <title>
         {% block title %}

tests/test_main.py

@@ -21,16 +21,15 @@ def test_web():
 
 def test_web_post(mocker):
     # Patch the get_db function to return the db mock
-    mocker.patch('src.link_liberate.database.Session_Local', return_value=Mock())
+    mocker.patch("src.link_liberate.database.Session_Local", return_value=Mock())
     client_mock = TestClient(app)
     response = client_mock.post("/liberate", data={"content": "valid_content"})
     assert response.status_code == 200
 
 
-
 def test_get_link(mocker):
     # Patch the get_db function to return the db mock
-    mocker.patch('src.link_liberate.database.Session_Local', return_value=Mock())
+    mocker.patch("src.link_liberate.database.Session_Local", return_value=Mock())
     client_mock = TestClient(app)
     response = client_mock.get("/valid_uuid", follow_redirects=False)
     assert response.status_code == 301