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 · master
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 c13a1aa Dawsh 2y ago from sqlalchemy.orm import Session
6 c13a1aa Dawsh 2y ago
7 f7af2b3 Dawsh 2y ago from app.db import crud, User, GetDB
8 667402d Erfan 1y ago from app.db.models import Service
9 c13a1aa Dawsh 2y ago from app.models.admin import Admin, oauth2_scheme
10 e652421 Dawsh 2y ago from app.utils.auth import get_admin_payload
11 c13a1aa Dawsh 2y ago
12 c13a1aa Dawsh 2y ago
13 f4fecee Dawsh 2y ago def get_db():
14 f7af2b3 Dawsh 2y ago with GetDB() as db:
15 f7af2b3 Dawsh 2y ago yield db
16 f7af2b3 Dawsh 2y ago
17 f7af2b3 Dawsh 2y ago
18 01cdf2e Dawsh 2y ago def get_admin(
19 01cdf2e Dawsh 2y ago db: Annotated[Session, Depends(get_db)],
20 01cdf2e Dawsh 2y ago token: Annotated[str, Depends(oauth2_scheme)],
21 01cdf2e Dawsh 2y ago ):
22 c13a1aa Dawsh 2y ago payload = get_admin_payload(token)
23 c13a1aa Dawsh 2y ago if not payload:
24 c13a1aa Dawsh 2y ago return
25 c13a1aa Dawsh 2y ago
26 01cdf2e Dawsh 2y ago dbadmin = crud.get_admin(db, payload["username"])
27 c13a1aa Dawsh 2y ago if not dbadmin:
28 c13a1aa Dawsh 2y ago return
29 c13a1aa Dawsh 2y ago
30 f9bc14b Dawsh 2y ago if dbadmin.password_reset_at:
31 f4fecee Dawsh 2y ago created_at = payload.get("created_at")
32 f4fecee Dawsh 2y ago if not created_at or dbadmin.password_reset_at > created_at:
33 c13a1aa Dawsh 2y ago return
34 c13a1aa Dawsh 2y ago
35 4708fcb Dawsh 2y ago if not dbadmin.is_sudo and not dbadmin.enabled:
36 4708fcb Dawsh 2y ago return
37 4708fcb 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 01cdf2e Dawsh 2y ago def get_subscription_user(
62 01cdf2e Dawsh 2y ago username: str, key: str, db: Annotated[Session, Depends(get_db)]
63 01cdf2e Dawsh 2y ago ):
64 c13a1aa Dawsh 2y ago try:
65 c13a1aa Dawsh 2y ago int(key, 16)
66 c13a1aa Dawsh 2y ago except ValueError:
67 f4fecee Dawsh 2y ago raise HTTPException(status_code=404)
68 c13a1aa Dawsh 2y ago
69 c13a1aa Dawsh 2y ago db_user = crud.get_user(db, username)
70 f4fecee Dawsh 2y ago if db_user and db_user.key == key:
71 f4fecee Dawsh 2y ago return db_user
72 f4fecee Dawsh 2y ago else:
73 f4fecee Dawsh 2y ago raise HTTPException(status_code=404)
74 c13a1aa Dawsh 2y ago
75 c13a1aa Dawsh 2y ago
76 01cdf2e Dawsh 2y ago def get_user(
77 01cdf2e Dawsh 2y ago username: str,
78 01cdf2e Dawsh 2y ago admin: Annotated[Admin, Depends(get_current_admin)],
79 01cdf2e Dawsh 2y ago db: Annotated[Session, Depends(get_db)],
80 01cdf2e Dawsh 2y ago ):
81 f7af2b3 Dawsh 2y ago db_user = crud.get_user(db, username)
82 01cdf2e Dawsh 2y ago if not (
83 f4fecee Dawsh 2y ago admin.is_sudo or (db_user and db_user.admin.username == admin.username)
84 01cdf2e Dawsh 2y ago ):
85 f7af2b3 Dawsh 2y ago raise HTTPException(status_code=403, detail="You're not allowed")
86 f7af2b3 Dawsh 2y ago
87 f7af2b3 Dawsh 2y ago if not db_user:
88 f7af2b3 Dawsh 2y ago raise HTTPException(status_code=404, detail="User not found")
89 f7af2b3 Dawsh 2y ago
90 f7af2b3 Dawsh 2y ago return db_user
91 f7af2b3 Dawsh 2y ago
92 f7af2b3 Dawsh 2y ago
93 4708fcb Dawsh 2y ago def user_modification_access(
94 07c8fa3 dawsh 1y ago admin: Annotated[Admin, Depends(get_current_admin)],
95 4708fcb Dawsh 2y ago ):
96 4708fcb Dawsh 2y ago if not admin.is_sudo and not admin.modify_users_access:
97 4708fcb Dawsh 2y ago raise HTTPException(status_code=403, detail="You're not allowed")
98 4708fcb Dawsh 2y ago
99 4708fcb Dawsh 2y ago
100 41ba8cb Dawsh 2y ago def parse_start_date(start: str | None = None):
101 41ba8cb Dawsh 2y ago if not start:
102 f1ab16c Dawsh 2y ago return datetime.fromtimestamp(
103 f9bc14b Dawsh 2y ago datetime.utcnow().timestamp() - 30 * 24 * 3600
104 f1ab16c Dawsh 2y ago )
105 41ba8cb Dawsh 2y ago else:
106 41ba8cb Dawsh 2y ago return datetime.fromisoformat(start)
107 41ba8cb Dawsh 2y ago
108 41ba8cb Dawsh 2y ago
109 41ba8cb Dawsh 2y ago def parse_end_date(end: str | None = None):
110 41ba8cb Dawsh 2y ago if not end:
111 f9bc14b Dawsh 2y ago return datetime.utcnow()
112 41ba8cb Dawsh 2y ago else:
113 41ba8cb Dawsh 2y ago return datetime.fromisoformat(end)
114 41ba8cb Dawsh 2y ago
115 41ba8cb Dawsh 2y ago
116 667402d Erfan 1y ago def get_service(id: int, db: Annotated[Session, Depends(get_db)]):
117 667402d Erfan 1y ago service = crud.get_service(db, id)
118 667402d Erfan 1y ago if not service:
119 667402d Erfan 1y ago raise HTTPException(status_code=404, detail="Service not found")
120 667402d Erfan 1y ago return service
121 667402d Erfan 1y ago
122 667402d Erfan 1y ago
123 f7af2b3 Dawsh 2y ago SubUserDep = Annotated[User, Depends(get_subscription_user)]
124 f7af2b3 Dawsh 2y ago UserDep = Annotated[User, Depends(get_user)]
125 c13a1aa Dawsh 2y ago AdminDep = Annotated[Admin, Depends(get_current_admin)]
126 c13a1aa Dawsh 2y ago SudoAdminDep = Annotated[Admin, Depends(sudo_admin)]
127 c13a1aa Dawsh 2y ago DBDep = Annotated[Session, Depends(get_db)]
128 41ba8cb Dawsh 2y ago StartDateDep = Annotated[datetime, Depends(parse_start_date)]
129 41ba8cb Dawsh 2y ago EndDateDep = Annotated[datetime, Depends(parse_end_date)]
130 4708fcb Dawsh 2y ago ModifyUsersAccess = Annotated[None, Depends(user_modification_access)]
131 667402d Erfan 1y ago ServiceDep = Annotated[Service, Depends(get_service)]