create cur user

src/api/dependacies/user_dep.py

@@ -0,0 +1,30 @@
+from typing import Annotated
+
+from fastapi import HTTPException, Depends
+from fastapi.security import OAuth2PasswordBearer
+from jwt import InvalidTokenError
+
+from src.core.auth_manager import AuthManager
+from src.core.settings import settings
+from src.schemas.auth import TokenData
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.api.v1_login_url}/login")
+
+
+async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
+    credentials_exception = HTTPException(
+        status_code=401,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = AuthManager.decode_access_token(token=token)
+        if payload is None:
+            raise credentials_exception
+        user = TokenData(**payload)
+    except InvalidTokenError:
+        raise credentials_exception
+    return user
+
+
+CurUsr = Annotated[TokenData, Depends(get_current_user)]

src/api/v1/users.py

@@ -1,5 +1,11 @@
 from fastapi import APIRouter
 
+from src.api.dependacies.user_dep import CurUsr
 from src.core.settings import settings
 
 router = APIRouter(prefix=settings.api.v1.users, tags=["Users"])
+
+
+@router.get("/me")
+async def get_me(user: CurUsr):
+    return {"user"}

src/core/auth_manager.py

@@ -33,3 +33,11 @@ class AuthManager:
             algorithm=settings.access_token.algorithm,
         )
         return encoded_jwt
+
+    @classmethod
+    def decode_access_token(cls, token: str) -> dict:
+        return jwt.decode(
+            token,
+            settings.access_token.secret_key,
+            algorithms=[settings.access_token.algorithm],
+        )

src/core/settings.py

@@ -12,11 +12,19 @@ class ApiV1Prefix(BaseModel):
     auth: str = "/auth"
     users: str = "/users"
 
+    @property
+    def login_url(self) -> str:
+        return f"{self.prefix}{self.auth}"
+
 
 class ApiPrefix(BaseModel):
     prefix: str = "/api"
     v1: ApiV1Prefix = ApiV1Prefix()
 
+    @property
+    def v1_login_url(self) -> str:
+        return f"{self.prefix}{self.v1.login_url}"
+
 
 class DbSettings(BaseModel):
     url: str = f"sqlite+aiosqlite:///{DB_PATH}"

src/schemas/auth.py

@@ -4,3 +4,8 @@ from pydantic import BaseModel
 class Token(BaseModel):
     access_token: str
     token_type: str
+
+
+class TokenData(BaseModel):
+    sub: str | None = None
+    is_active: bool

src/services/auth.py

@@ -34,5 +34,7 @@ class AuthService(BaseService):
                 status_code=401,
                 detail="Incorrect username or password",
             )
-        access_token = AuthManager.create_access_token(data={"sub": user.username})
+        access_token = AuthManager.create_access_token(
+            data={"sub": user.username, "is_active": user.is_active}
+        )
         return Token(access_token=access_token, token_type="bearer")