change(assist-stats): auth

assist-stats/auth.py

@@ -0,0 +1,33 @@
+from fastapi.security import OAuth2PasswordBearer
+from fastapi import HTTPException, Depends, status
+from decouple import config
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+
+class AuthHandler:
+    def __init__(self):
+        """
+        Authorization method using an API key.
+        """
+        self.__api_keys = [config("ACCESS_TOKEN")]
+
+    def __contains__(self, api_key):
+        return api_key in self.__api_keys
+
+    def add_key(self, key):
+        """Adds new key for authentication."""
+        self.__api_keys.append(key)
+
+
+auth_method = AuthHandler()
+
+
+def api_key_auth(api_key: str = Depends(oauth2_scheme)):
+    """Method to verify auth."""
+    global auth_method
+    if api_key not in auth_method:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Forbidden"
+        )

assist-stats/env.default

@@ -8,3 +8,5 @@ POOL_SIZE=20
 MAX_OVERFLOW=10
 POOL_TIMEOUT=30
 POOL_RECYCLE=3600
+
+ACCESS_TOKEN=abc

assist-stats/main.py

@@ -9,6 +9,7 @@ from sqlalchemy import Enum
 from sqlalchemy import CheckConstraint
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import sessionmaker, Session
+from auth import api_key_auth
 
 pg_dbname = config("pg_dbname")
 pg_host = config("pg_host")
@@ -147,7 +148,7 @@ def insert_event(event: EventCreate, db: Session):
         db.close()
 
 
-@app.post("/events")
+@app.post("/assist-stats/events", dependencies=[Depends(api_key_auth)])
 def create_event(event: EventCreate, db: Session = Depends(get_db)):
     if event.event_state == EventStateEnum.end:
         update_duration(event.event_id, event.timestamp, db)