Merge pull request #58 from PythonFloripa/feature/#33

Feature/#33

Author: dan5e3s6ares

.vscode/settings.json

@@ -9,5 +9,10 @@
     },
     "python-envs.defaultEnvManager": "ms-python.python:poetry",
     "python-envs.defaultPackageManager": "ms-python.python:poetry",
-    "python-envs.pythonProjects": []
+    "python-envs.pythonProjects": [],
+    "python.testing.pytestArgs": [
+      "tests"
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
 }

app/main.py

@@ -4,6 +4,7 @@
 from fastapi import FastAPI
 from slowapi import _rate_limit_exceeded_handler
 
+from app.routers.admin.routes import create_admin
 from app.routers.router import setup_router as setup_router_v2
 from app.services.database.database import AsyncSessionLocal, init_db
 from app.services.limiter import limiter
@@ -16,6 +17,7 @@ async def lifespan(app: FastAPI):
     # add check db file and create if not found
     await init_db()
     app.db_session_factory = AsyncSessionLocal()
+    await create_admin(app.db_session_factory)
     try:
         yield
     finally:

app/routers/admin/__init__.py

@@ -0,0 +1,65 @@
+import os
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Request, status
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from app.routers.authentication import get_current_active_community
+from app.schemas import CommunityPostResponse
+from app.services import auth
+from app.services.database.models import Community as DBCommunity  # Precisa?
+from app.services.database.orm.community import create_community
+from app.services.limiter import limiter
+
+# ADMIN_USER = os.getenv("ADMIN_USER")
+# ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD")
+
+
+async def create_admin(session: AsyncSession):
+    ADMIN_USER = os.getenv("ADMIN_USER")
+    ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD")
+    ADMIN_EMAIL = os.getenv("ADMIN_EMAIL")
+    password = ADMIN_PASSWORD
+    hashed_password = auth.hash_password(password)
+    community = DBCommunity(
+        username=ADMIN_USER,
+        email=ADMIN_EMAIL,
+        password=hashed_password,
+        role="admin",
+    )
+    await create_community(session=session, community=community)
+
+    return {"msg": "Admin successfully created"}
+
+
+def setup():
+    router = APIRouter(prefix="/admin", tags=["admin"])
+
+    @router.post(
+        "/create_community",
+        response_model=CommunityPostResponse,
+        status_code=status.HTTP_201_CREATED,
+        summary="Create Community endpoint",
+        description="Create Community and returns a confirmation message",
+    )
+    @limiter.limit("60/minute")
+    async def post_create_community(
+        request: Request,
+        admin_community: Annotated[
+            DBCommunity, Depends(get_current_active_community)
+        ],
+        community: DBCommunity,
+    ):
+        """
+        Server Admin endpoint that creates Community and returns a confirmation
+          message.
+        """
+        admin_role = admin_community.role
+        if admin_role != "admin":
+            return {"status": "Unauthorized"}
+        session: AsyncSession = request.app.db_session_factory
+        await create_community(session=session, community=community)
+
+        return CommunityPostResponse()
+
+    return router

app/routers/admin/routes.py

@@ -70,25 +70,6 @@ async def authenticate_community(
             return None
         return found_community
 
-    # Teste
-
-    @router.post("/create_commumity")
-    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,
-        )
-        session: AsyncSession = request.app.db_session_factory
-        session.add(community)
-        await session.commit()
-        await session.refresh(community)
-        return {"msg": "succes? "}
-
-    # Teste
-
     @router.post("/token", response_model=Token)
     @limiter.limit("60/minute")
     async def login_for_access_token(

app/routers/authentication.py

@@ -1,5 +1,6 @@
 from fastapi import APIRouter
 
+from app.routers.admin.routes import setup as admin_router_setup
 from app.routers.authentication import setup as authentication_router_setup
 from app.routers.healthcheck.routes import setup as healthcheck_router_setup
 from app.routers.libraries.routes import setup as libraries_router_setup
@@ -12,4 +13,5 @@ def setup_router() -> APIRouter:
     router.include_router(news_router_setup(), prefix="")
     router.include_router(authentication_router_setup(), prefix="")
     router.include_router(libraries_router_setup(), prefix="")
+    router.include_router(admin_router_setup(), prefix="")
     return router

app/routers/router.py

@@ -34,9 +34,14 @@ class Community(BaseModel):
     email: str
 
 
-# Extends Community Class with hashed password
+# Extends Community Class with hashed password and role
 class CommunityInDB(Community):
     password: str
+    role: str
+
+
+class CommunityPostResponse(BaseModel):
+    status: str = "Community Criado"
 
 
 class News(BaseModel):

app/schemas.py

@@ -1,6 +1,7 @@
 from datetime import datetime
 from typing import Optional
 
+from sqlalchemy import Text
 from sqlmodel import Field, SQLModel
 
 
@@ -9,10 +10,11 @@ class Community(SQLModel, table=True):
 
     id: Optional[int] = Field(default=None, primary_key=True)
     username: str
-    email: str
+    email: str = Field(sa_column=Text)  # VARCHAR(255)
     password: str
     created_at: Optional[datetime] = Field(default_factory=datetime.now)
     updated_at: Optional[datetime] = Field(
         default_factory=datetime.now,
         sa_column_kwargs={"onupdate": datetime.now},
     )
+    role: str = Field(default="user")  # user or admin

app/services/database/models/communities.py

@@ -4,6 +4,7 @@
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from app.services.database.models import Community
+from app.services.encryption import decrypt_email, encrypt_email
 
 
 async def get_community_by_username(
@@ -19,6 +20,29 @@ async def get_community_by_username(
 
     # Executa a declaração na sessão e retorna o primeiro resultado
     result = await session.exec(statement)
-    community = result.first()
+    community_result = result.first()
+
+    # add tratamento de descriptografia do email
+    if community_result is not None:
+        # evitar mutação direta e cache
+        community = community_result.model_copy()
+        community.email = decrypt_email(community.email)
+    else:
+        community = None
+    return community  # add community not found treatment?
+
+
+async def create_community(
+    session: AsyncSession,
+    community: Community,  # community model
+) -> Optional[Community]:
+    """
+    Cria um novo membro da comunidade.
+    Somente usuário autenticado e com role Admin podem executar.
+    """
+    community.email = encrypt_email(community.email)
+    session.add(community)
+    await session.commit()
+    await session.refresh(community)
 
     return community

app/services/database/orm/community.py

@@ -0,0 +1,27 @@
+import os
+
+from cryptography.fernet import Fernet
+
+test_encryption_key = "r0-QKv5qACJNFRqy2cNZCsfZ_zVvehlC-v8zDJb--EI="
+# print(Fernet.generate_key())
+# Carrega a chave da variável de ambiente
+ENCRYPTION_KEY = os.getenv("ENCRYPTION_KEY", test_encryption_key)
+
+if not ENCRYPTION_KEY:
+    raise RuntimeError(
+        "ENCRYPTION_KEY não está definida nas variáveis de ambiente."
+    )
+
+cipher = Fernet(ENCRYPTION_KEY)
+
+
+def encrypt_email(email: str) -> str:
+    """Criptografa uma string de e-mail."""
+    encrypted_email = cipher.encrypt(email.encode())
+    return encrypted_email
+
+
+def decrypt_email(encrypted_email: str) -> str:
+    """Descriptografa uma string de e-mail."""
+    decrypted_email = cipher.decrypt(encrypted_email).decode()
+    return decrypted_email