gitscope
/
← marzneshin / app/dependencies.py / blame
Files Commits 446 Branches 1 Tags 43 Markers 4 README License
https://dev.kronoz.win/r/marzneshin.git
app/dependencies.py · v0.1.4
Source
1 41ba8cb Dawsh 2y ago from datetime import datetime
2 c13a1aa Dawsh 2y ago from typing import Annotated
3 c13a1aa Dawsh 2y ago
4 c13a1aa Dawsh 2y ago from fastapi import Depends, HTTPException, status
5 f7af2b3 Dawsh 2y ago from sqlalchemy.exc import SQLAlchemyError
6 c13a1aa Dawsh 2y ago from sqlalchemy.orm import Session
7 c13a1aa Dawsh 2y ago
8 f7af2b3 Dawsh 2y ago from app.db import crud, User, GetDB
9 c13a1aa Dawsh 2y ago from app.models.admin import Admin, oauth2_scheme
10 c13a1aa Dawsh 2y ago from app.utils.jwt import get_admin_payload
11 c13a1aa Dawsh 2y ago from config import SUDOERS
12 c13a1aa Dawsh 2y ago
13 c13a1aa Dawsh 2y ago
14 f7af2b3 Dawsh 2y ago def get_db(): # Dependency
15 f7af2b3 Dawsh 2y ago with GetDB() as db:
16 f7af2b3 Dawsh 2y ago yield db
17 f7af2b3 Dawsh 2y ago
18 f7af2b3 Dawsh 2y ago
19 c13a1aa Dawsh 2y ago def get_admin(db: Annotated[Session, Depends(get_db)],
20 c13a1aa Dawsh 2y ago token: Annotated[str, Depends(oauth2_scheme)]):
21 c13a1aa Dawsh 2y ago payload = get_admin_payload(token)
22 c13a1aa Dawsh 2y ago if not payload:
23 c13a1aa Dawsh 2y ago return
24 c13a1aa Dawsh 2y ago
25 c13a1aa Dawsh 2y ago if payload['username'] in SUDOERS and payload['is_sudo'] is True:
26 c13a1aa Dawsh 2y ago return Admin(username=payload['username'], is_sudo=True)
27 c13a1aa Dawsh 2y ago
28 c13a1aa Dawsh 2y ago dbadmin = crud.get_admin(db, payload['username'])
29 c13a1aa Dawsh 2y ago if not dbadmin:
30 c13a1aa Dawsh 2y ago return
31 c13a1aa Dawsh 2y ago
32 c13a1aa Dawsh 2y ago if dbadmin.password_reset_at:
33 c13a1aa Dawsh 2y ago if not payload.get("created_at"):
34 c13a1aa Dawsh 2y ago return
35 c13a1aa Dawsh 2y ago if dbadmin.password_reset_at > payload.get("created_at"):
36 c13a1aa Dawsh 2y ago return
37 c13a1aa Dawsh 2y ago
38 c13a1aa Dawsh 2y ago return Admin.model_validate(dbadmin)
39 c13a1aa Dawsh 2y ago
40 c13a1aa Dawsh 2y ago
41 c13a1aa Dawsh 2y ago def get_current_admin(admin: Annotated[Admin, Depends(get_admin)]):
42 c13a1aa Dawsh 2y ago if not admin:
43 c13a1aa Dawsh 2y ago raise HTTPException(
44 c13a1aa Dawsh 2y ago status_code=status.HTTP_401_UNAUTHORIZED,
45 c13a1aa Dawsh 2y ago detail="Could not validate credentials",
46 c13a1aa Dawsh 2y ago headers={"WWW-Authenticate": "Bearer"},
47 c13a1aa Dawsh 2y ago )
48 c13a1aa Dawsh 2y ago
49 c13a1aa Dawsh 2y ago return admin
50 c13a1aa Dawsh 2y ago
51 c13a1aa Dawsh 2y ago
52 c13a1aa Dawsh 2y ago def sudo_admin(admin: Annotated[Admin, Depends(get_current_admin)]):
53 c13a1aa Dawsh 2y ago if not admin.is_sudo:
54 c13a1aa Dawsh 2y ago raise HTTPException(
55 c13a1aa Dawsh 2y ago status_code=status.HTTP_403_FORBIDDEN,
56 c13a1aa Dawsh 2y ago detail="Access Denied",
57 c13a1aa Dawsh 2y ago )
58 f7af2b3 Dawsh 2y ago return admin
59 c13a1aa Dawsh 2y ago
60 c13a1aa Dawsh 2y ago
61 f7af2b3 Dawsh 2y ago def get_subscription_user(username: str, key: str, db: Annotated[Session, Depends(get_db)]):
62 c13a1aa Dawsh 2y ago try:
63 c13a1aa Dawsh 2y ago int(key, 16)
64 c13a1aa Dawsh 2y ago except ValueError:
65 c13a1aa Dawsh 2y ago raise HTTPException(404)
66 c13a1aa Dawsh 2y ago
67 c13a1aa Dawsh 2y ago db_user = crud.get_user(db, username)
68 c13a1aa Dawsh 2y ago if not db_user or db_user.key != key:
69 c13a1aa Dawsh 2y ago raise HTTPException(404)
70 c13a1aa Dawsh 2y ago return db_user
71 c13a1aa Dawsh 2y ago
72 c13a1aa Dawsh 2y ago
73 f7af2b3 Dawsh 2y ago def get_user(username: str,
74 f7af2b3 Dawsh 2y ago admin: Annotated[Admin, Depends(get_current_admin)],
75 f7af2b3 Dawsh 2y ago db: Annotated[Session, Depends(get_db)]):
76 f7af2b3 Dawsh 2y ago db_user = crud.get_user(db, username)
77 f7af2b3 Dawsh 2y ago if not (admin.is_sudo or (db_user.admin and db_user.admin.username == admin.username)):
78 f7af2b3 Dawsh 2y ago raise HTTPException(status_code=403, detail="You're not allowed")
79 f7af2b3 Dawsh 2y ago
80 f7af2b3 Dawsh 2y ago if not db_user:
81 f7af2b3 Dawsh 2y ago raise HTTPException(status_code=404, detail="User not found")
82 f7af2b3 Dawsh 2y ago
83 f7af2b3 Dawsh 2y ago return db_user
84 f7af2b3 Dawsh 2y ago
85 f7af2b3 Dawsh 2y ago
86 41ba8cb Dawsh 2y ago def parse_start_date(start: str | None = None):
87 41ba8cb Dawsh 2y ago if not start:
88 41ba8cb Dawsh 2y ago return datetime.fromtimestamp(datetime.utcnow().timestamp() - 30 * 24 * 3600)
89 41ba8cb Dawsh 2y ago else:
90 41ba8cb Dawsh 2y ago return datetime.fromisoformat(start)
91 41ba8cb Dawsh 2y ago
92 41ba8cb Dawsh 2y ago
93 41ba8cb Dawsh 2y ago def parse_end_date(end: str | None = None):
94 41ba8cb Dawsh 2y ago if not end:
95 41ba8cb Dawsh 2y ago return datetime.utcnow()
96 41ba8cb Dawsh 2y ago else:
97 41ba8cb Dawsh 2y ago return datetime.fromisoformat(end)
98 41ba8cb Dawsh 2y ago
99 41ba8cb Dawsh 2y ago
100 f7af2b3 Dawsh 2y ago SubUserDep = Annotated[User, Depends(get_subscription_user)]
101 f7af2b3 Dawsh 2y ago UserDep = Annotated[User, Depends(get_user)]
102 c13a1aa Dawsh 2y ago AdminDep = Annotated[Admin, Depends(get_current_admin)]
103 c13a1aa Dawsh 2y ago SudoAdminDep = Annotated[Admin, Depends(sudo_admin)]
104 c13a1aa Dawsh 2y ago DBDep = Annotated[Session, Depends(get_db)]
105 41ba8cb Dawsh 2y ago StartDateDep = Annotated[datetime, Depends(parse_start_date)]
106 41ba8cb Dawsh 2y ago EndDateDep = Annotated[datetime, Depends(parse_end_date)]