feature: add get_current_community()

refactor: reorganize imports and improve authentication logic in routers

Author: IkeSalmonson

app/routers/authentication.py

@@ -1,58 +1,119 @@
-from fastapi import APIRouter, Depends, HTTPException, status, Request
-from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
-from sqlmodel.ext.asyncio.session import AsyncSession
+from typing import Annotated
+
 import jwt
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
 from jwt.exceptions import InvalidTokenError
+from sqlmodel.ext.asyncio.session import AsyncSession
 
+from app.schemas import Token, TokenPayload
 from app.services import auth
-from app.schemas import Token, TokenPayload, Community
 from app.services.database.models import Community as DBCommunity
 from app.services.database.orm.community import get_community_by_username
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/authentication/token")
 
+
 def setup():
-    router = APIRouter(prefix='/authentication', tags=['authentication'])
-    async def authenticate_community( request: Request , username: str, password: str):
-       # Valida se o usuário existe e se a senha está correta
-       session: AsyncSession = request.app.db_session_factory
-       found_community = await get_community_by_username(
-                                    username=username,
-                                    session= session
-                                    )
-       if not found_community or not auth.verify_password(password, found_community.password):
+    router = APIRouter(prefix="/authentication", tags=["authentication"])
+
+    async def authenticate_community(
+        request: Request, username: str, password: str
+    ):
+        # Valida se o usuário existe e se a senha está correta
+        session: AsyncSession = request.app.db_session_factory
+        found_community = await get_community_by_username(
+            username=username, session=session
+        )
+        if not found_community or not auth.verify_password(
+            password, found_community.password
+        ):
             return None
-       return found_community
+        return found_community
 
+    async def get_current_community(
+        request: Request,
+        token: Annotated[str, Depends(oauth2_scheme)],
+    ) -> DBCommunity:
+        credentials_exception = HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Could not validate credentials",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
 
-    #### Teste 
+        try:
+            payload = jwt.decode(
+                token, auth.SECRET_KEY, algorithms=[auth.ALGORITHM]
+            )
+            username = payload.get("sub")
+            if username is None:
+                raise credentials_exception
+            token_data = TokenPayload(username=username)
+        except InvalidTokenError:
+            raise credentials_exception
+        session: AsyncSession = request.app.db_session_factory
+        community = await get_community_by_username(
+            session=session, username=token_data.username
+        )
+        if community is None:
+            raise credentials_exception
+
+        return community
+
+    async def get_current_active_community(
+        current_user: Annotated[DBCommunity, Depends(get_current_community)],
+    ) -> DBCommunity:
+        # A função simplesmente retorna o usuário.
+        # Pode ser estendido futuramente para verificar um status "ativo".
+        return current_user
+
+    # Teste
 
     @router.post("/create_commumity")
-    async def create_community(request: Request ):
+    async def create_community(request: Request):
         password = "123Asd!@#"
-        hashed_password=auth.hash_password(password)
-        community = DBCommunity(username="username", email="username@test.com", password=hashed_password)
+        hashed_password = auth.hash_password(password)
+        community = DBCommunity(
+            username="username",
+            email="username@test.com",
+            password=hashed_password,
+        )
         session: AsyncSession = request.app.db_session_factory
         session.add(community)
         await session.commit()
         await session.refresh(community)
-        return {'msg':'succes? '}
-    #### Teste 
+        return {"msg": "succes? "}
+
+    # Teste
 
     @router.post("/token", response_model=Token)
-    async def login_for_access_token(request: Request , form_data: OAuth2PasswordRequestForm = Depends() ) :
+    async def login_for_access_token(
+        request: Request, form_data: OAuth2PasswordRequestForm = Depends()
+    ):
         # Rota de login: valida credenciais e retorna token JWT
-        community = await authenticate_community( request, form_data.username, form_data.password)
+        community = await authenticate_community(
+            request, form_data.username, form_data.password
+        )
         if not community:
             raise HTTPException(
                 status_code=status.HTTP_401_UNAUTHORIZED,
-                detail="Credenciais inválidas"
+                detail="Credenciais inválidas",
             )
         payload = TokenPayload(username=community.username)
         token, expires_in = auth.create_access_token(data=payload)
         return {
             "access_token": token,
             "token_type": "Bearer",
-            "expires_in": expires_in
+            "expires_in": expires_in,
         }
-    return router # Retorna o router configurado com as rotas de autenticação
+
+    @router.get("/me", response_model=DBCommunity)
+    async def read_community_me(
+        current_community: Annotated[
+            DBCommunity, Depends(get_current_active_community)
+        ],
+    ):
+        # Rota para obter informações do usuário autenticado
+        return current_community
+
+    return router  # Retorna o router configurado com as rotas de autenticação

app/routers/libraries/routes.py

@@ -1,9 +1,9 @@
 from fastapi import APIRouter, Request, status
 from pydantic import BaseModel
-from services.database.orm.library import insert_library
 
 from app.schemas import Library as LibrarySchema
 from app.services.database.models.libraries import Library
+from app.services.database.orm.library import insert_library
 
 
 class LibraryResponse(BaseModel):

app/services/auth.py

@@ -1,41 +1,36 @@
-#from passlib.context import CryptContext
-import bcrypt
+# from passlib.context import CryptContext
+import os
 from datetime import datetime, timedelta, timezone
-from app.schemas import TokenPayload
+
+import bcrypt
 import jwt
-import os
+
+from app.schemas import TokenPayload
 
 SECRET_KEY = os.getenv("SECRET_KEY", "default_fallback_key")
 ALGORITHM = os.getenv("ALGORITHM", "HS256")
 ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", 20))
 
-def verify_password(plain, hashed): 
+
+def verify_password(plain, hashed):
     # Verifica se a senha passada bate com a hash da comunidade
     return bcrypt.checkpw(
         bytes(plain, encoding="utf-8"),
         hashed,
     )
 
-def hash_password(password): 
+
+def hash_password(password):
     # Retorna a senha em hash para salvar no banco de dados
     return bcrypt.hashpw(
         bytes(password, encoding="utf-8"),
         bcrypt.gensalt(),
     )
 
 
-
-#pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
-
-#def verify_password(plain, hashed): 
-#    # Verifica se a senha passada bate com a hash da comunidade
-#    return pwd_context.verify(plain, hashed)
-#
-#def hash_password(password): 
-#    # Retorna a senha em hash para salvar no banco de dados
-#    return pwd_context.hash(password)
-
-def create_access_token(data: TokenPayload, expires_delta: timedelta | None = None):
+def create_access_token(
+    data: TokenPayload, expires_delta: timedelta | None = None
+):
     """
     Gera um token JWT contendo os dados do usuário (payload) e uma data de expiração.
     JWT specification says that there's a key sub (subject) that should be used to identify the user.

app/services/database/orm/community.py

@@ -1,21 +1,24 @@
 from typing import Optional
+
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
+
 from app.services.database.models import Community
 
 
 async def get_community_by_username(
     username: str,
-    session: AsyncSession,) -> Optional[Community]:
+    session: AsyncSession,
+) -> Optional[Community]:
     """
     Busca e retorna um membro da comunidade pelo nome de usuário.
     Retorna None se o usuário não for encontrado.
     """
     # Cria a declaração SQL para buscar a comunidade pelo nome de usuário
     statement = select(Community).where(Community.username == username)
-    
+
     # Executa a declaração na sessão e retorna o primeiro resultado
     result = await session.exec(statement)
     community = result.first()
-    
-    return community
+
+    return community

tests/test_subscriptions.py

@@ -1,11 +1,9 @@
 import pytest
 import pytest_asyncio
-
+from services.database.models import Community, Subscription
 from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
-from services.database.models import Community, Subscription
-
 
 @pytest_asyncio.fixture
 async def community(session: AsyncSession):
@@ -26,7 +24,9 @@ async def test_insert_subscription(session: AsyncSession, community: Community):
     session.add(subscription)
     await session.commit()
 
-    statement = select(Subscription).where(Subscription.email == "teste@teste.com")
+    statement = select(Subscription).where(
+        Subscription.email == "teste@teste.com"
+    )
     result = await session.exec(statement)
     found = result.first()