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.3.2
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 c13a1aa Dawsh 2y ago from app.models.admin import Admin, oauth2_scheme
9 e652421 Dawsh 2y ago from app.utils.auth import get_admin_payload
10 c13a1aa Dawsh 2y ago
11 c13a1aa Dawsh 2y ago
12 f4fecee Dawsh 2y ago def get_db():
13 f7af2b3 Dawsh 2y ago with GetDB() as db:
14 f7af2b3 Dawsh 2y ago yield db
15 f7af2b3 Dawsh 2y ago
16 f7af2b3 Dawsh 2y ago
17 01cdf2e Dawsh 2y ago def get_admin(
18 01cdf2e Dawsh 2y ago db: Annotated[Session, Depends(get_db)],
19 01cdf2e Dawsh 2y ago token: Annotated[str, Depends(oauth2_scheme)],
20 01cdf2e Dawsh 2y ago ):
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 01cdf2e Dawsh 2y ago dbadmin = crud.get_admin(db, payload["username"])
26 c13a1aa Dawsh 2y ago if not dbadmin:
27 c13a1aa Dawsh 2y ago return
28 c13a1aa Dawsh 2y ago
29 f9bc14b Dawsh 2y ago if dbadmin.password_reset_at:
30 f4fecee Dawsh 2y ago created_at = payload.get("created_at")
31 f4fecee Dawsh 2y ago if not created_at or dbadmin.password_reset_at > created_at:
32 c13a1aa Dawsh 2y ago return
33 c13a1aa Dawsh 2y ago
34 c13a1aa Dawsh 2y ago return Admin.model_validate(dbadmin)
35 c13a1aa Dawsh 2y ago
36 c13a1aa Dawsh 2y ago
37 c13a1aa Dawsh 2y ago def get_current_admin(admin: Annotated[Admin, Depends(get_admin)]):
38 c13a1aa Dawsh 2y ago if not admin:
39 c13a1aa Dawsh 2y ago raise HTTPException(
40 c13a1aa Dawsh 2y ago status_code=status.HTTP_401_UNAUTHORIZED,
41 c13a1aa Dawsh 2y ago detail="Could not validate credentials",
42 c13a1aa Dawsh 2y ago headers={"WWW-Authenticate": "Bearer"},
43 c13a1aa Dawsh 2y ago )
44 c13a1aa Dawsh 2y ago
45 c13a1aa Dawsh 2y ago return admin
46 c13a1aa Dawsh 2y ago
47 c13a1aa Dawsh 2y ago
48 c13a1aa Dawsh 2y ago def sudo_admin(admin: Annotated[Admin, Depends(get_current_admin)]):
49 c13a1aa Dawsh 2y ago if not admin.is_sudo:
50 c13a1aa Dawsh 2y ago raise HTTPException(
51 c13a1aa Dawsh 2y ago status_code=status.HTTP_403_FORBIDDEN,
52 c13a1aa Dawsh 2y ago detail="Access Denied",
53 c13a1aa Dawsh 2y ago )
54 f7af2b3 Dawsh 2y ago return admin
55 c13a1aa Dawsh 2y ago
56 c13a1aa Dawsh 2y ago
57 01cdf2e Dawsh 2y ago def get_subscription_user(
58 01cdf2e Dawsh 2y ago username: str, key: str, db: Annotated[Session, Depends(get_db)]
59 01cdf2e Dawsh 2y ago ):
60 c13a1aa Dawsh 2y ago try:
61 c13a1aa Dawsh 2y ago int(key, 16)
62 c13a1aa Dawsh 2y ago except ValueError:
63 f4fecee Dawsh 2y ago raise HTTPException(status_code=404)
64 c13a1aa Dawsh 2y ago
65 c13a1aa Dawsh 2y ago db_user = crud.get_user(db, username)
66 f4fecee Dawsh 2y ago if db_user and db_user.key == key:
67 f4fecee Dawsh 2y ago return db_user
68 f4fecee Dawsh 2y ago else:
69 f4fecee Dawsh 2y ago raise HTTPException(status_code=404)
70 c13a1aa Dawsh 2y ago
71 c13a1aa Dawsh 2y ago
72 01cdf2e Dawsh 2y ago def get_user(
73 01cdf2e Dawsh 2y ago username: str,
74 01cdf2e Dawsh 2y ago admin: Annotated[Admin, Depends(get_current_admin)],
75 01cdf2e Dawsh 2y ago db: Annotated[Session, Depends(get_db)],
76 01cdf2e Dawsh 2y ago ):
77 f7af2b3 Dawsh 2y ago db_user = crud.get_user(db, username)
78 01cdf2e Dawsh 2y ago if not (
79 f4fecee Dawsh 2y ago admin.is_sudo or (db_user and db_user.admin.username == admin.username)
80 01cdf2e Dawsh 2y ago ):
81 f7af2b3 Dawsh 2y ago raise HTTPException(status_code=403, detail="You're not allowed")
82 f7af2b3 Dawsh 2y ago
83 f7af2b3 Dawsh 2y ago if not db_user:
84 f7af2b3 Dawsh 2y ago raise HTTPException(status_code=404, detail="User not found")
85 f7af2b3 Dawsh 2y ago
86 f7af2b3 Dawsh 2y ago return db_user
87 f7af2b3 Dawsh 2y ago
88 f7af2b3 Dawsh 2y ago
89 41ba8cb Dawsh 2y ago def parse_start_date(start: str | None = None):
90 41ba8cb Dawsh 2y ago if not start:
91 f1ab16c Dawsh 2y ago return datetime.fromtimestamp(
92 f9bc14b Dawsh 2y ago datetime.utcnow().timestamp() - 30 * 24 * 3600
93 f1ab16c Dawsh 2y ago )
94 41ba8cb Dawsh 2y ago else:
95 41ba8cb Dawsh 2y ago return datetime.fromisoformat(start)
96 41ba8cb Dawsh 2y ago
97 41ba8cb Dawsh 2y ago
98 41ba8cb Dawsh 2y ago def parse_end_date(end: str | None = None):
99 41ba8cb Dawsh 2y ago if not end:
100 f9bc14b Dawsh 2y ago return datetime.utcnow()
101 41ba8cb Dawsh 2y ago else:
102 41ba8cb Dawsh 2y ago return datetime.fromisoformat(end)
103 41ba8cb Dawsh 2y ago
104 41ba8cb Dawsh 2y ago
105 f7af2b3 Dawsh 2y ago SubUserDep = Annotated[User, Depends(get_subscription_user)]
106 f7af2b3 Dawsh 2y ago UserDep = Annotated[User, Depends(get_user)]
107 c13a1aa Dawsh 2y ago AdminDep = Annotated[Admin, Depends(get_current_admin)]
108 c13a1aa Dawsh 2y ago SudoAdminDep = Annotated[Admin, Depends(sudo_admin)]
109 c13a1aa Dawsh 2y ago DBDep = Annotated[Session, Depends(get_db)]
110 41ba8cb Dawsh 2y ago StartDateDep = Annotated[datetime, Depends(parse_start_date)]
111 41ba8cb Dawsh 2y ago EndDateDep = Annotated[datetime, Depends(parse_end_date)]