Lock Information and Maintenance Done!!

brewman/main.py

@@ -15,6 +15,8 @@ from .routers import (
     fingerprint,
     issue,
     issue_grid,
+    lock_information,
+    maintenance,
     payment,
     product,
     product_group,
@@ -112,6 +114,9 @@ app.include_router(purchase.router, prefix="/api/purchase", tags=["vouchers"])
 app.include_router(purchase_return.router, prefix="/api/purchase-return", tags=["vouchers"])
 app.include_router(issue.router, prefix="/api/issue", tags=["vouchers"])
 
+app.include_router(lock_information.router, prefix="/api/lock-information", tags=["settings"])
+app.include_router(maintenance.router, prefix="/api/maintenance", tags=["settings"])
+
 
 def init():
     uvicorn.run(app, host=Settings.host, port=Settings.port)

brewman/routers/lock_information.py

@@ -0,0 +1,105 @@
+from fastapi import (
+    APIRouter,
+    Depends,
+    Security,
+)
+from sqlalchemy.orm import Session
+
+from ..models.master import DbSetting
+from ..schemas.auth import UserToken
+from ..core.security import get_current_active_user as get_user
+from ..db.session import SessionLocal
+from brewman.schemas.settings import LockInformation
+
+router = APIRouter()
+
+
+# Dependency
+def get_db() -> Session:
+    try:
+        db = SessionLocal()
+        yield db
+    finally:
+        db.close()
+
+
+@router.post("", response_model=LockInformation)
+def post(
+    data_in: LockInformation,
+    db: Session = Depends(get_db),
+    user: UserToken = Security(get_user, scopes=["lock-date"]),
+):
+    data = {
+        "Start": {"Locked": data_in.lock_older},
+        "Finish": {"Locked": data_in.lock_newer},
+    }
+
+    if data_in.lock_older:
+        data["Start"]["Rolling"] = data_in.older_rolling
+        if data_in.older_rolling:
+            data["Start"]["Days"] = data_in.older_days
+        else:
+            data["Start"]["Date"] = data_in.older_date
+    if data_in.lock_newer:
+        data["Finish"]["Rolling"] = data_in.newer_rolling
+        if data_in.newer_rolling:
+            data["Finish"]["Days"] = data_in.newer_days
+        else:
+            data["Finish"]["Date"] = data_in.newer_date
+
+    lock_date = db.query(DbSetting).filter(DbSetting.name == "Lock Info").first()
+    if lock_date is not None:
+        lock_date.data = data
+    else:
+        lock_date = DbSetting(name="Lock Info", data=data)
+        db.add(lock_date)
+    db.commit()
+    return get_info(data)
+
+
+@router.delete("")
+def delete(
+    db: Session = Depends(get_db),
+    user: UserToken = Security(get_user, scopes=["lock-date"]),
+):
+    lock_date = db.query(DbSetting).filter(DbSetting.name == "Lock Info").first()
+    if lock_date is not None and lock_date.data is not None:
+        lock_date.data = None
+    db.commit()
+    return get_info({})
+
+
+@router.get("")
+def get(
+    db: Session = Depends(get_db), user: UserToken = Security(get_user),
+):
+    data = db.query(DbSetting).filter(DbSetting.name == "Lock Info").first()
+    if data is None:
+        return get_info({})
+    else:
+        return get_info(data.data)
+
+
+def get_info(data):
+    info = {}
+    if "Start" not in data:
+        info["lockOlder"] = False
+    else:
+        info["lockOlder"] = data["Start"]["Locked"]
+        if data["Start"]["Locked"]:
+            info["olderRolling"] = data["Start"]["Rolling"]
+            if data["Start"]["Rolling"]:
+                info["olderDays"] = data["Start"]["Days"]
+            else:
+                info["olderDate"] = data["Start"]["Date"].strftime("%d-%b-%Y")
+    if "Finish" not in data:
+        info["lockNewer"] = False
+    else:
+        info["lockNewer"] = data["Finish"]["Locked"]
+        if data["Finish"]["Locked"]:
+            info["newerRolling"] = data["Finish"]["Rolling"]
+            if data["Finish"]["Rolling"]:
+                info["newerDays"] = data["Finish"]["Days"]
+            else:
+                info["newerDate"] = data["Finish"]["Date"].strftime("%d-%b-%Y")
+    return info

brewman/routers/maintenance.py

@@ -0,0 +1,58 @@
+from fastapi import (
+    APIRouter,
+    Depends,
+    Security,
+)
+from sqlalchemy.orm import Session
+
+from ..models import User
+from ..models.master import DbSetting
+from ..schemas.auth import UserToken
+from ..core.security import get_current_active_user as get_user
+from ..db.session import SessionLocal
+from brewman.schemas.settings import Maintenance
+
+router = APIRouter()
+
+
+# Dependency
+def get_db() -> Session:
+    try:
+        db = SessionLocal()
+        yield db
+    finally:
+        db.close()
+
+
+@router.get("", response_model=Maintenance)
+def get_maintenance(
+    db: Session = Depends(get_db), user: UserToken = Security(get_user),
+):
+    data = db.query(DbSetting).filter(DbSetting.name == "Maintenance").first()
+    return info(data, db)
+
+
+@router.post("", response_model=Maintenance)
+def set_maintenance(
+    data: Maintenance,
+    db: Session = Depends(get_db),
+    user: UserToken = Security(get_user, scopes=["maintenance"]),
+):
+    maintenance = db.query(DbSetting).filter(DbSetting.name == "Maintenance").first()
+    if data.enabled is False and maintenance is not None:
+        db.delete(maintenance)
+        maintenance = None
+    elif data.enabled is True and maintenance is None:
+        maintenance = DbSetting(name="Maintenance", data=user.id_)
+        db.add(maintenance)
+    elif data.enabled is True and maintenance.data != user.id_:
+        maintenance.data = user.id_
+    db.commit()
+    return info(maintenance, db)
+
+
+def info(data, db: Session):
+    if data is None:
+        return {"enabled": False, "user": ""}
+    user = db.query(User).filter(User.id == data.data).one()
+    return {"enabled": True, "user": user.name}

brewman/routers/management/settings.py

@@ -1,108 +0,0 @@
-import datetime
-from brewman.models.auth import User
-from brewman.models.master import DbSetting
-
-
-@router.post("/api/lock-information")  # "Lock Date"
-def set_lock_info(request):
-    start_locked = request.json_body["lockOlder"]
-    finish_locked = request.json_body["lockNewer"]
-    data = {"Start": {"Locked": start_locked}, "Finish": {"Locked": finish_locked}}
-
-    if start_locked:
-        rolling = request.json_body["olderRolling"]
-        data["Start"]["Rolling"] = rolling
-        if rolling:
-            data["Start"]["Days"] = request.json_body["olderDays"]
-        else:
-            data["Start"]["Date"] = datetime.datetime.strptime(
-                request.json_body["olderDate"], "%d-%b-%Y"
-            )
-    if finish_locked:
-        rolling = request.json_body["newerRolling"]
-        data["Finish"]["Rolling"] = rolling
-        if rolling:
-            data["Finish"]["Days"] = request.json_body["newerDays"]
-        else:
-            data["Finish"]["Date"] = datetime.datetime.strptime(
-                request.json_body["newerDate"], "%d-%b-%Y"
-            )
-
-    lock_date = (
-        request.dbsession.query(DbSetting).filter(DbSetting.name == "Lock Info").first()
-    )
-    if lock_date is not None:
-        lock_date.data = data
-    else:
-        lock_date = DbSetting(name="Lock Info", data=data)
-        request.dbsession.add(lock_date)
-    transaction.commit()
-    return get_lock_info(request)
-
-
-@router.delete("/api/lock-information")  # "Lock Date"
-def clear_lock_info(request):
-    lock_date = (
-        request.dbsession.query(DbSetting).filter(DbSetting.name == "Lock Info").first()
-    )
-    if lock_date is not None and lock_date.data is not None:
-        lock_date.data = None
-    transaction.commit()
-    return {}
-
-
-@router.get("/api/lock-information")  # "Authenticated"
-def get_lock_info(request):
-    data = (
-        request.dbsession.query(DbSetting).filter(DbSetting.name == "Lock Info").first()
-    )
-    if data is None:
-        return {"lockOlder": False, "lockNewer": False}
-    data = data.data
-    info = {"lockOlder": data["Start"]["Locked"], "lockNewer": data["Finish"]["Locked"]}
-    if data["Start"]["Locked"]:
-        info["olderRolling"] = data["Start"]["Rolling"]
-        if data["Start"]["Rolling"]:
-            info["olderDays"] = data["Start"]["Days"]
-        else:
-            info["olderDate"] = data["Start"]["Date"].strftime("%d-%b-%Y")
-
-    if data["Finish"]["Locked"]:
-        info["newerRolling"] = data["Finish"]["Rolling"]
-        if data["Finish"]["Rolling"]:
-            info["newerDays"] = data["Finish"]["Days"]
-        else:
-            info["newerDate"] = data["Finish"]["Date"].strftime("%d-%b-%Y")
-    return info
-
-
-@router.get("/api/maintenance")  # "Authenticated"
-def get_maintenance(request):
-    data = (
-        request.dbsession.query(DbSetting)
-        .filter(DbSetting.name == "Maintenance")
-        .first()
-    )
-    if data is None:
-        return {"enabled": False, "user": ""}
-    user = request.dbsession.query(User).filter(User.id == data.data).one()
-    return {"enabled": True, "user": user.name}
-
-
-@router.post("/api/maintenance")  # "Maintenance"
-def set_maintenance(request):
-    status = request.json_body["enabled"]
-    maintenance = (
-        request.dbsession.query(DbSetting)
-        .filter(DbSetting.name == "Maintenance")
-        .first()
-    )
-    if status is False and maintenance is not None:
-        request.dbsession.delete(maintenance)
-    elif status is True and maintenance is None:
-        maintenance = DbSetting(name="Maintenance", data=request.authenticated_userid)
-        request.dbsession.add(maintenance)
-    elif status is True and maintenance.data != request.authenticated_userid:
-        maintenance.data = request.authenticated_userid
-    transaction.commit()
-    return get_maintenance(request)

brewman/schemas/settings.py

@@ -0,0 +1,40 @@
+import uuid
+from typing import List, Optional
+from datetime import datetime, date
+from pydantic import BaseModel, validator
+
+from brewman.schemas import to_camel
+
+
+class LockInformation(BaseModel):
+    lock_older: bool
+    lock_newer: bool
+    older_rolling: Optional[bool]
+    older_days: Optional[int]
+    older_date: Optional[date]
+    newer_rolling: Optional[bool]
+    newer_days: Optional[int]
+    newer_date: Optional[date]
+
+    class Config:
+        alias_generator = to_camel
+        json_encoders = {
+            date: lambda v: v.strftime("%d-%b-%Y"),
+        }
+
+    @validator("older_date", pre=True)
+    def parse_older_date(cls, value):
+        if isinstance(value, date):
+            return value
+        return datetime.strptime(value, "%d-%b-%Y").date()
+
+    @validator("newer_date", pre=True)
+    def parse_newer_date(cls, value):
+        if isinstance(value, date):
+            return value
+        return datetime.strptime(value, "%d-%b-%Y").date()
+
+
+class Maintenance(BaseModel):
+    enabled: bool
+    user: Optional[str]