fix

.deploy/deploy-dev.yaml

@@ -6,6 +6,21 @@ services:
     image: mathwave/sprint-repo:locks
     networks:
       - locks-development
+    command: api
+    environment:
+      MONGO_HOST: "mongo.develop.sprinthub.ru"
+      MONGO_PASSWORD: $MONGO_PASSWORD_DEV
+    deploy:
+      mode: replicated
+      restart_policy:
+        condition: any
+      update_config:
+        parallelism: 1
+        order: start-first
+
+  cleaner:
+    image: mathwave/sprint-repo:locks
+    command: cleaner
     environment:
       MONGO_HOST: "mongo.develop.sprinthub.ru"
       MONGO_PASSWORD: $MONGO_PASSWORD_DEV

.deploy/deploy-prod.yaml

@@ -6,6 +6,21 @@ services:
     image: mathwave/sprint-repo:locks
     networks:
       - locks
+    command: api
+    environment:
+      MONGO_HOST: "mongo.sprinthub.ru"
+      MONGO_PASSWORD: $MONGO_PASSWORD_PROD
+    deploy:
+      mode: replicated
+      restart_policy:
+        condition: any
+      update_config:
+        parallelism: 1
+        order: start-first
+
+  cleaner:
+    image: mathwave/sprint-repo:locks
+    command: cleaner
     environment:
       MONGO_HOST: "mongo.sprinthub.ru"
       MONGO_PASSWORD: $MONGO_PASSWORD_PROD

app/daemons/api.py

@@ -0,0 +1,20 @@
+import fastapi
+import uvicorn
+
+from app.routers import acquire
+from app.routers import release
+
+from app.storage import mongo
+from daemons import base
+
+
+class Daemon(base.Daemon):
+    def execute(self):
+        app = fastapi.FastAPI()
+
+        app.include_router(acquire.router)
+        app.include_router(release.router)
+
+        mongo.create_indexes()
+
+        uvicorn.run(app, host="0.0.0.0", port=80)

app/daemons/base.py

@@ -0,0 +1,3 @@
+class Daemon:
+    def execute(self):
+        raise NotImplementedError

app/daemons/cleaner.py

@@ -0,0 +1,11 @@
+from app.daemons import base
+from app.storage.mongo import locks
+import time
+from app.utils import time as time_utils
+
+
+class Daemon(base.Daemon):
+    def execute(self):
+        while True:
+            locks.collection.delete_many({'locked_until': {'$lt': time_utils.now()}})
+            time.sleep(5)

app/routers/acquire.py

@@ -1,9 +1,7 @@
-import datetime
 import fastapi
 import pydantic
 
 from app.storage.mongo import locks
-from app.utils import time
 
 
 class RequestBody(pydantic.BaseModel):
@@ -17,7 +15,6 @@ router = fastapi.APIRouter()
 @router.post('/api/v1/acquire', status_code=fastapi.status.HTTP_202_ACCEPTED, responses={'409': {'description': 'Conflict'}})
 async def execute(body: RequestBody):
     try:
-        await locks.acquire(locks.Lock(name=body.name, locked_until=time.now() + datetime.timedelta(seconds=body.ttl)))
-    except Exception as e:
-        print(e)
+        await locks.acquire(body.name, ttl=body.ttl)
+    except locks.ConflictException:
         raise fastapi.HTTPException(409)

app/routers/release.py

@@ -16,4 +16,4 @@ router = fastapi.APIRouter()
 
 @router.post('/api/v1/release', status_code=fastapi.status.HTTP_202_ACCEPTED)
 async def execute(body: RequestBody):
-    await locks.release(locks.Lock(name=body.name, locked_until=time.now() + datetime.timedelta(seconds=body.ttl)))
+    await locks.release(body.name)

app/storage/mongo/__init__.py

@@ -18,3 +18,9 @@ def create_indexes():
     database.get_collection('locks').create_index([
         ('name', 1), 
     ], unique=True)
+
+
+def remove_outdated():
+    client = pymongo.MongoClient(CONNECTION_STRING)
+    while True:
+        

app/storage/mongo/locks.py

@@ -1,22 +1,25 @@
-import bson
 import datetime
-import pydantic
 
 from app.storage.mongo import database
 from app.utils import time
 from bson import codec_options
 
+from pymongo import errors
+
+
+class ConflictException(Exception):
+    pass
+
 
 collection = database.get_collection("locks", codec_options=codec_options.CodecOptions(tz_aware=True))
 
 
-class Lock(pydantic.BaseModel):
-    name: str
-    locked_until: pydantic.AwareDatetime
-
-
-async def acquire(lock: Lock):
-    await collection.insert_one(lock.model_dump())
+async def acquire(name: str, ttl: int):
+    locked_until = time.now() + datetime.timedelta(seconds=ttl)
+    try:
+        await collection.insert_one({'name': name, 'locked_until': locked_until})
+    except errors.DuplicateKeyError:
+        raise ConflictException
 
 
 async def release(name: str):

main.py

@@ -1,19 +1,12 @@
-import fastapi
-import uvicorn
-
-from app.routers import acquire
-from app.routers import release
-
-from app.storage import mongo
+import sys
 
 
-app = fastapi.FastAPI()
+arg = sys.argv[-1]
+if arg == 'api':
+    from app.daemons.api import Daemon
+elif arg == 'cleaner':
+    from app.daemons.cleaner import Daemon
+else:
+    raise NotImplementedError('Daemon is not implemented')
 
-app.include_router(acquire.router)
-app.include_router(release.router)
-
-mongo.create_indexes()
-
-
-if __name__ == '__main__':
-    uvicorn.run(app, host="0.0.0.0", port=80)
+Daemon().execute()