refactor: enhance authentication flow with async session management and add tests for token endpoint

Author: IkeSalmonson

app/main.py

@@ -32,7 +32,6 @@ async def get_db_session():
     async with app.db_session_factory() as session:
         yield session
 
-
-app.include_router(setup_router_v2(), prefix="/api")
+app.include_router(setup_router_v2(get_db_session), prefix="/api") 
 
 logger.info("PyNews Server Starter")

app/routers/authentication.py

@@ -1,59 +1,56 @@
 from fastapi import APIRouter, Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
-from app.services import auth
-from app.services.database.community import get_community_by_username # Atualizar após banco de dados
-from app.schemas import Token, TokenPayload, Community
+from sqlmodel.ext.asyncio.session import AsyncSession
 import jwt
 from jwt.exceptions import InvalidTokenError
 
+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.community import get_community_by_username
+
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/authentication/token")
 
-def setup():
+def setup(get_db_session_dep):
     router = APIRouter(prefix='/authentication', tags=['authentication'])
-
-    async def authenticate_community(username: str, password: str):
+    async def authenticate_community(username: str, password: str, session: AsyncSession = Depends(get_db_session_dep)):
         # Valida se o usuário existe e se a senha está correta
-        db_user = await get_community_by_username(username)
-        if not db_user or not auth.verify_password(password, db_user.password):
+        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 db_user
+        return found_community
+
+
+    #### Teste 
+
+    @router.post("/create_commumity")
+    async def create_community( session: AsyncSession = Depends(get_db_session_dep)):
+        password = "123Asd!@#"
+        hashed_password=auth.hash_password(password)
+        community = DBCommunity(username="username", email="username@test.com", password=hashed_password)
+        session.add(community)
+        await session.commit()
+        await session.refresh(community)
+        return {'msg':'succes? '}
+    #### Teste 
 
     @router.post("/token", response_model=Token)
-    async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
+    async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), session: AsyncSession = Depends(get_db_session_dep)) :
         # Rota de login: valida credenciais e retorna token JWT
-        community = await authenticate_community(form_data.username, form_data.password)
+        community = await authenticate_community(form_data.username, form_data.password, session)
         if not community:
             raise HTTPException(
                 status_code=status.HTTP_401_UNAUTHORIZED,
                 detail="Credenciais inválidas"
             )
-        # Community ex: email='alice@example.com' id=1 username='alice' full_name="Alice in the Maravilha's world" password='$2b$12$cA3fzLrRCmLp1aKn6ULhF.sQfaPQ70EoJU3Q0Szf6e4/YaVsKAAHS'
         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
         }
-
-
-    @router.get("/me", response_model=Community)
-    async def get_current_community(token: str = Depends(oauth2_scheme)):
-        # Rota protegida: retorna dados do usuário atual com base no token
-        creds_exc = HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Token inválido",
-            headers={"WWW-Authenticate": "Bearer"},
-        )
-        try:
-            payload_dict = jwt.decode(token, auth.SECRET_KEY, algorithms=[auth.ALGORITHM])
-            payload = TokenPayload(**payload_dict)
-        except (  InvalidTokenError ,ValueError):
-            raise creds_exc
-
-        community = get_community_by_username(payload.sub)
-        if not community:
-            raise creds_exc
-        return community
-
     return router # Retorna o router configurado com as rotas de autenticação

app/routers/router.py

@@ -4,9 +4,9 @@
 from app.routers.news.routes import setup as news_router_setup
 from app.routers.authentication import setup as authentication_router_setup
 
-def setup_router() -> APIRouter:
+def setup_router(get_db_session_dep) -> APIRouter:
     router = APIRouter()
     router.include_router(healthcheck_router_setup(), prefix="")
     router.include_router(news_router_setup(), prefix="")
-    router.include_router(authentication_router_setup(), prefix='')
+    router.include_router(authentication_router_setup(get_db_session_dep), prefix='')
     return router

app/schemas.py

@@ -3,13 +3,24 @@
 from typing import List
 from enum import Enum
 
+## News
+class News(BaseModel):
+    description: str
+    tag: str
+
+class Library(BaseModel):
+    library_name: str
+    news: list[News]
+    logo: HttpUrl
+    version: str
+    release_date: datetime
+    release_doc_url: HttpUrl
 
-## User Class
+## Community / User Class
 class Community(BaseModel):
     username: str
-    full_name: str
     email: str
-
+## Extends Community Class with hashed password
 class CommunityInDB(Community):
     password: str
 
@@ -32,17 +43,3 @@ class TagEnum(str, Enum):
 class Subscription(BaseModel):
     tags: List[TagEnum]
     libraries_list: List[str]
-
-## News
-class News(BaseModel):
-    description: str
-    tag: str
-
-
-class Library(BaseModel):
-    library_name: str
-    news: list[News]
-    logo: HttpUrl
-    version: str
-    release_date: datetime
-    release_doc_url: HttpUrl

app/services/auth.py

@@ -12,17 +12,16 @@
 
 def verify_password(plain, hashed): 
     # Verifica se a senha passada bate com a hash da comunidade
-    print(plain, hashed)
     return pwd_context.verify(plain, hashed)
 
-def get_password_hash(password): 
+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):
     """
     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.
     Parâmetros:
     - data (TokenPayload): Dicionário com os dados que serão codificados no token. Deve conter a chave 'sub' com o identificador do usuário.
     - expires_delta (timedelta | None): Tempo até o token expirar. Se não fornecido, usará o padrão de 20 minutos.

app/services/database/community.py

@@ -1,12 +1,22 @@
-# app/services/database/community.py
+from typing import Optional
 from sqlmodel import select
-from sqlalchemy.exc import NoResultFound
-from app.services.database.model.community_model import Community
-from app.services.database.database import get_session
+from sqlmodel.ext.asyncio.session import AsyncSession
+from app.services.database.models import Community
 
-async def get_community_by_username(username: str):
-    async for session in get_session():
-        stmt = select(Community).where(Community.username == username)
-        result = await session.exec(stmt)
-        user = result.one_or_none()
-        return user
+
+async def get_community_by_username(
+    username: str,
+    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

app/services/database/database.py

@@ -32,7 +32,7 @@
     engine,
     class_=AsyncSession,
     expire_on_commit=False,
-    echo=True,  # expire_on_commit=False é importante!
+    #echo=True,  # expire_on_commit=False é importante!
 )