Models: SafeUser/SafePrintJob, PrintJobLog redesign, job_id→id, datetime timestamps, model-based request.state.user

decorators/auth.py

@@ -48,10 +48,10 @@ async def wrapper(*args, **kwargs):
             if user == "not_found":
                 raise HTTPException(status_code=404, detail="User not found")
 
-            if user["suspended"]:
+            if user.suspended:
                 raise HTTPException(status_code=403, detail="User suspended")
 
-            if not user["admin"] and require_admin:
+            if not user.admin and require_admin:
                 raise HTTPException(status_code=403, detail="Admin required")
 
             request.state.user = user

models/__init__.py

@@ -0,0 +1,4 @@
+from models.user import User, SafeUser, UserPrinterSettings, UserPermissions
+from models.print_job import PrintJob, SafePrintJob, PrintJobLog
+
+__all__ = ["User", "SafeUser", "UserPrinterSettings", "UserPermissions", "PrintJob", "SafePrintJob", "PrintJobLog"]

models/print_job.py

@@ -0,0 +1,65 @@
+from pydantic import BaseModel, Field, model_validator
+from typing import Optional
+from datetime import datetime
+import hashlib
+
+
+class PrintJobLog(BaseModel):
+    id: str
+    timestamp: datetime
+    actor: str  # "system" or a user id
+    type: str   # e.g. "job_created", "job_accepted", "job_rejected"
+    description: Optional[str] = None
+
+    @model_validator(mode="before")
+    @classmethod
+    def _coerce_old_format(cls, data):
+        # Backward compatibility: old logs had {timestamp, job_id, user_id, description}.
+        if isinstance(data, dict) and "job_id" in data and "id" not in data:
+            data = dict(data)
+            raw = f"{data.get('job_id', '')}{data.get('timestamp', '')}"
+            data["id"] = hashlib.md5(raw.encode()).hexdigest()
+            data["actor"] = data.pop("user_id", "system")
+            data["type"] = "legacy"
+            data.pop("job_id", None)
+        return data
+
+
+class PrintJob(BaseModel):
+    id: str
+    user_id: str
+    cups_job_id: Optional[str] = None
+    filename: str
+    file: str
+    color: bool = True
+    copies: int = 1
+    status: str
+    logs: list[PrintJobLog] = Field(default_factory=list)
+    created_at: datetime
+    updated_at: datetime
+
+    @model_validator(mode="before")
+    @classmethod
+    def _coerce_old_format(cls, data):
+        # Backward compatibility: old documents stored the job identifier as "job_id".
+        if isinstance(data, dict) and "job_id" in data and "id" not in data:
+            data = dict(data)
+            data["id"] = data.pop("job_id")
+        return data
+
+    def to_safe(self) -> "SafePrintJob":
+        return SafePrintJob.model_validate(self.model_dump())
+
+
+class SafePrintJob(BaseModel):
+    """PrintJob with internal/sensitive fields redacted (no cups_job_id, no file path)."""
+
+    id: str
+    user_id: str
+    filename: str
+    color: bool = True
+    copies: int = 1
+    status: str
+    logs: list[PrintJobLog] = Field(default_factory=list)
+    created_at: datetime
+    updated_at: datetime

models/user.py

@@ -0,0 +1,47 @@
+from pydantic import BaseModel, Field, model_validator
+from typing import Optional
+from datetime import datetime
+
+
+class UserPrinterSettings(BaseModel):
+    credits: float = 0
+    no_credits_action: str = "require_approval"
+
+
+class UserPermissions(BaseModel):
+    manage_printer: bool = False
+    manage_users: bool = False
+
+
+class SafeUser(BaseModel):
+    """User with personal/sensitive information redacted (no password)."""
+
+    id: str
+    name: str
+    email: str
+    auth_methods: list[str] = Field(default_factory=list)
+    region: str
+    language: str
+    superadmin: bool = False
+    admin: bool = False
+    printer: UserPrinterSettings = Field(default_factory=UserPrinterSettings)
+    permissions: UserPermissions = Field(default_factory=UserPermissions)
+    suspended: bool = False
+    created_at: datetime
+    updated_at: datetime
+
+    @model_validator(mode="before")
+    @classmethod
+    def _coerce_permissions(cls, data):
+        # Backward compatibility: existing users stored permissions as a list.
+        # Discard the list and use default UserPermissions values instead.
+        if isinstance(data, dict) and isinstance(data.get("permissions"), list):
+            data = dict(data)
+            data["permissions"] = UserPermissions()
+        return data
+
+
+class User(SafeUser):
+    """Full user model including the hashed password."""
+
+    password: Optional[str] = None

routers/authentication.py

@@ -25,7 +25,7 @@ async def get_auth_methods(request: Request):
 
     return {
         "success": True,
-        "methods": user["auth_methods"]
+        "methods": user.auth_methods
     }
 
 @router.post("/otp/send")
@@ -37,16 +37,16 @@ async def send_otp(request: Request):
     if user == "not_found":
         return {"success": False, "message": "User not found"}
 
-    if not "otp" in user["auth_methods"]:
+    if not "otp" in user.auth_methods:
         return {"success": False, "message": "OTP authentication is disabled."}
 
-    rate_limited = login_codes._is_rate_limited(user["id"])
+    rate_limited = login_codes._is_rate_limited(user.id)
     if rate_limited:
         return {"success": False, "message": "Too many requests. Please try again later."}
 
-    attempt_id, random_code = login_codes.generate_otp(user["id"])
+    attempt_id, random_code = login_codes.generate_otp(user.id)
 
-    if user["language"] == "PT":
+    if user.language == "PT":
         subject = "O seu código de autenticação"
         template = "auth_otp_login_pt"
     else:
@@ -56,7 +56,7 @@ async def send_otp(request: Request):
     m = mailer.send_email(
         sender=os.environ.get("DEFAULT_SENDER_EMAIL"),
         sender_name=os.environ.get("DEFAULT_SENDER_NAME"),
-        to=user["email"],
+        to=user.email,
         subject=subject,
         template=template,
         otp=random_code
@@ -90,14 +90,14 @@ async def password_login(request: Request):
     if user == "not_found":
         return {"success": False, "message": "User not found"}
 
-    if not "password" in user["auth_methods"]:
+    if not "password" in user.auth_methods:
         return {"success": False, "message": "Password authentication is disabled."}
 
-    verify = user_tools.verify_password_hash(password, user["password"])
+    verify = user_tools.verify_password_hash(password, user.password)
     if not verify:
         return {"success": False, "message": "Invalid password."}
 
-    token = sessions_controller.create_session(user["id"])
+    token = sessions_controller.create_session(user.id)
 
     return {"success": True, "token": token}
 

routers/printer.py

@@ -9,14 +9,14 @@
 @require_auth()
 async def get_print_jobs(request: Request, page: int = 1):
 
-    jobs = printer.get_user_jobs(request.state.user["id"], page=page)
+    jobs = printer.get_user_jobs(request.state.user.id, page=page)
     return {"success": True, "data": jobs or []}
 
 @router.get("/jobs/{id}")
 @require_auth()
 async def get_print_job(request: Request, id: str):
     job = printer.get_job(id)
-    if not job or job["user_id"] != request.state.user["id"]:
+    if not job or job["user_id"] != request.state.user.id:
         return {"success": False, "message": "Job not found"}
 
     return {"success": True, "data": job}

tasks/proccess_printer_email.py

@@ -75,8 +75,10 @@ def proccess_printer_email(request):
             language="EN"
         )
         user = user_tools.get_user_by_email(ai_response["email"])
+        if user == "not_found":
+            return
 
-    if user["suspended"]:
+    if user.suspended:
         return
 
     valid_files = ai_response["files"]
@@ -97,20 +99,20 @@ def proccess_printer_email(request):
         total_cost += printer.calculate_cost(file["content"], color=file["color"], copies=file["copies"])
 
 
-    if user["printer"]["credits"] < total_cost and user["printer"]["no_credits_action"] == "require_approval":
+    if user.printer.credits < total_cost and user.printer.no_credits_action == "require_approval":
         mailer.send_email(
             sender_name="Brick Printer",
             sender_email=os.environ.get("PRINTER_EMAIL"),
-            to=user["email"],
+            to=user.email,
             subject="Your print job is on hold",
             template="not_enough_credits_en",
-            name=user["name"],
-            credits=user["printer"]["credits"],
+            name=user.name,
+            credits=user.printer.credits,
             total_cost=total_cost
         )
         set_pending = True
         return
-    elif user["printer"]["credits"] < total_cost and user["printer"]["no_credits_action"] != "require_approval":
+    elif user.printer.credits < total_cost and user.printer.no_credits_action != "require_approval":
         # This should never be happening. Adding this for a future implementation.
         return
 
@@ -121,7 +123,7 @@ def proccess_printer_email(request):
         file_str = f"printer_jobs/{str(uuid.uuid4())}.pdf"
         storage.upload_file(os.environ.get("S3_USER_CONTENT_BUCKET"), file_str, base64.b64decode(file["content"]))
         job = {
-            "user_id": user["id"],
+            "user_id": user.id,
             "filename": file["name"],
             "file": file_str,
             "color": file["color"],

tools/printer.py

@@ -1,5 +1,6 @@
 from utils.database import Database
 from tools.users import UserTools
+from models.print_job import PrintJob, PrintJobLog
 import datetime
 import uuid
 from utils.printer import Printer as PrinterConnector
@@ -27,56 +28,55 @@ def calculate_cost(self, file, color=True, copies=1):
 
         return cost
 
-    def _create_log(self, job_id, user_id, message):
-        now_ts = datetime.datetime.now(datetime.timezone.utc).timestamp()
-        log_entry = {
-            "timestamp": now_ts,
-            "job_id": job_id,
-            "user_id": user_id,
-            "description": message
-        }
-        return log_entry
+    def _create_log(self, actor, log_type, description=None):
+        now_ts = datetime.datetime.now(datetime.timezone.utc)
+        return PrintJobLog(
+            id=str(uuid.uuid4()),
+            timestamp=now_ts,
+            actor=actor,
+            type=log_type,
+            description=description,
+        )
 
     def register_job(self, user_id, filename, file, color=True, copies=1, status="pending"):
         job_id = str(uuid.uuid4())
-        now_ts = datetime.datetime.now(datetime.timezone.utc).timestamp()
-        creation_log = self._create_log(job_id=job_id, user_id=user_id, message="Created this print job.")
+        now_ts = datetime.datetime.now(datetime.timezone.utc)
+        creation_log = self._create_log(actor=user_id, log_type="job_created", description="Created this print job.")
 
-        self.db.mongo.print_jobs_v2.insert_one({
-            "_id": job_id,
-            "job_id": job_id,
-            "user_id": user_id,
-            "cups_job_id": None,
-            "filename": filename,
-            "file": file,
-            "color": color,
-            "copies": copies,
-            "status": status,
-            "logs": [creation_log],
-            "created_at": now_ts,
-            "updated_at": now_ts
-        })
+        job = PrintJob(
+            id=job_id,
+            user_id=user_id,
+            filename=filename,
+            file=file,
+            color=color,
+            copies=copies,
+            status=status,
+            logs=[creation_log],
+            created_at=now_ts,
+            updated_at=now_ts,
+        )
+
+        job_dict = job.model_dump()
+        job_dict["_id"] = job_id  # MongoDB requires _id
+
+        self.db.mongo.print_jobs_v2.insert_one(job_dict)
 
     def register_bulk_jobs(self, jobs):
         for job in jobs:
             self.register_job(job["user_id"], job["filename"], job["file"], color=job["color"], copies=job["copies"], status=job["status"])
 
     def get_user_jobs(self, user, page=1, per_page=10):
         skip = (page - 1) * per_page
-        jobs = list(self.db.mongo.print_jobs_v2.find({"user_id": user}).sort("created_at", -1).skip(skip).limit(per_page))
-        for job in jobs:
-            del job["_id"]
-        return jobs
+        raw_jobs = list(self.db.mongo.print_jobs_v2.find({"user_id": user}).sort("created_at", -1).skip(skip).limit(per_page))
+        return [PrintJob.model_validate(j).to_safe().model_dump() for j in raw_jobs]
 
     def admin_get_pending_jobs(self, page=1, per_page=10):
         skip = (page - 1) * per_page
-        jobs = list(self.db.mongo.print_jobs_v2.find({"status": "pending"}).sort("created_at", -1).skip(skip).limit(per_page))
-        for job in jobs:
-            del job["_id"]
-        return jobs
+        raw_jobs = list(self.db.mongo.print_jobs_v2.find({"status": "pending"}).sort("created_at", -1).skip(skip).limit(per_page))
+        return [PrintJob.model_validate(j).model_dump() for j in raw_jobs]
 
     def get_job(self, id):
-        job = self.db.mongo.print_jobs_v2.find_one({"job_id": id})
-        if job:
-            del job["_id"]
-        return job
+        raw = self.db.mongo.print_jobs_v2.find_one({"_id": id})
+        if not raw:
+            return None
+        return PrintJob.model_validate(raw).to_safe().model_dump()