Instalar Git
Instalar Python
Instalar PostgreSQL
Instalar o IDE da sua Escolha
Clonar o Projeto
git clone https://rinkon.unicv.cv/sti-desenvolvimento/siga-api.git
cd siga-api
Criar um Ambiente Virtual
python -m venv .venv
Ativar o Ambiente Virtual
source .venv\Scripts\activate
source .venv/bin/activate
Instalar Requisitos do Projeto
git checkout dev
pip install -r dev_requirements.txt
Configurar Variáveis de Ambiente
.env no diretório raiz do projeto usando .env.example como base:cp .env.example .env
.env e preencha os valores de configuração necessários para o seu ambiente de desenvolvimento.# para comentar quaisquer variáveis de ambiente para as quais não tenha valores disponíveis. Isto previne erros durante a inicialização da aplicação..env:DB_HOST=localhost
DB_PORT=5432
DB_USER=seu_usuario_db
DB_PASSWORD=sua_senha_db
DB_NAME=nome_sua_db
Aplicar Migrações Alembic
alembic upgrade head
uvicorn main:app --reload
http://localhost:8000 no seu navegador web.Este guia orienta-o na criação de um novo módulo chamado example com um único modelo e endpoints CRUD básicos, seguindo as convenções da API SIGA.
Crie a seguinte estrutura de pastas em app/modules/example/:
app/modules/example/
├── config/
│ ├── __init__.py
│ ├── routes.py
│ └── alembic.py
├── controllers/
│ ├── __init__.py
│ └── example_controller.py
├── models/
│ ├── __init__.py
│ └── example_model.py
├── schemas/
│ ├── __init__.py
│ └── example_schema.py
├── services/
│ ├── __init__.py
│ └── example_service.py
└── __init__.py
Crie app/modules/example/models/example_model.py:
from sqlalchemy import Column, String, Integer
from app.core.database import Base
from app.modules.basic.models.log_model import Log
class Example(Base, Log):
__tablename__ = "exam_example"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(255), nullable=False, index=True)
description = Column(String(500), nullable=True)
Pontos-Chave:
exam_ para o módulo example)Base e LogLog fornece campos de auditoria: user_id, ip_address, timestamp, statusCrie app/modules/example/schemas/example_schema.py:
from pydantic import BaseModel
from typing import Optional
from datetime import datetime
class ExampleRequest(BaseModel):
name: str
description: Optional[str] = None
class ExampleResponse(BaseModel):
id: int
name: str
description: Optional[str]
class Config:
from_attributes = True
Pontos-Chave:
ExampleRequest para dados de entradaExampleResponse para dados de saída (inclui campos de auditoria)from_attributes = True para compatibilidade ORMCrie app/modules/example/services/example_service.py:
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from ..models.example_model import Example
from ..schemas.example_schema import ExampleRequest
class ExampleService:
@staticmethod
async def create_example(db: AsyncSession, example_data: ExampleRequest, log_audit_info: dict) -> Example:
db_example = Example(
name=example_data.name,
description=example_data.description,
user_id=log_audit_info.get("user_id"),
ip_address=log_audit_info.get("ip_address")
)
db.add(db_example)
await db.commit()
await db.refresh(db_example)
return db_example
@staticmethod
async def get_example_by_id(db: AsyncSession, example_id: int) -> Example:
result = await db.execute(select(Example).where(Example.id == example_id).where(Example.status == 'ACTIVE'))
return result.scalars().first()
@staticmethod
async def get_all_examples(db: AsyncSession, limit: int, offset: int):
result = await db.execute(select(Example).where(Example.status == 'ACTIVE').limit(limit).offset(offset))
return result.scalars().all()
@staticmethod
async def update_example(db: AsyncSession, example_id: int, example_data: ExampleRequest, log_audit_info: dict) -> Example:
db_example = await ExampleService.get_example_by_id(db, example_id)
if db_example:
db_example.name = example_data.name
db_example.description = example_data.description
db_example.user_id = log_audit_info.get("user_id")
db_example.ip_address = log_audit_info.get("ip_address")
await db.commit()
await db.refresh(db_example)
return db_example
@staticmethod
async def delete_example(db: AsyncSession, example_id: int) -> Example:
db_example = await ExampleService.get_example_by_id(db, example_id)
if db_example:
db_example.soft_delete()
await db.commit()
return db_example
Pontos-Chave:
async e @staticmethodlog_audit_info para operações de criação/atualização/exclusãosoft_delete() ao invés de exclusões definitivasCrie app/modules/example/controllers/example_controller.py:
from fastapi import APIRouter, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.modules.basic.services.auth_service import validate_token, get_log_audit_info
from app.modules.basic.decorators.permission_decorator import require_permission
from app.modules.basic.schemas.auth import UniversalResponse
from ..schemas.example_schema import ExampleRequest, ExampleResponse
from ..services.example_service import ExampleService
router = APIRouter()
@router.post("/", response_model=UniversalResponse, summary="Criar Exemplo")
@require_permission("/v1/example/example/create")
async def create_example(
example_data: ExampleRequest,
token_data: dict = Depends(validate_token),
db: AsyncSession = Depends(get_db),
log_audit_info: dict = Depends(get_log_audit_info)
):
"""
Cria um novo exemplo no sistema.
Args:
example_data: Dados do exemplo a ser criado
token_data: Dados do token de autenticação
db: Sessão da base de dados
log_audit_info: Informações de auditoria
Returns:
UniversalResponse com o exemplo criado
"""
result = await ExampleService.create_example(db, example_data, log_audit_info)
return UniversalResponse.success_response(payload=ExampleResponse.model_validate(result))
@router.get("/{exampleId}", response_model=UniversalResponse, summary="Obter Exemplo por ID")
@require_permission("/v1/example/example/list")
async def get_example(
exampleId: int,
token_data: dict = Depends(validate_token),
db: AsyncSession = Depends(get_db)
):
"""
Obtém um exemplo pelo ID.
Args:
exampleId: ID do exemplo
token_data: Dados do token de autenticação
db: Sessão da base de dados
Returns:
UniversalResponse com o exemplo encontrado
"""
result = await ExampleService.get_example_by_id(db, exampleId)
if not result:
return UniversalResponse.error_response(
code=404,
cause="Not Found",
message="Exemplo não encontrado"
)
return UniversalResponse.success_response(payload=ExampleResponse.model_validate(result))
@router.get("/", response_model=UniversalResponse, summary="Listar Todos os Exemplos")
@require_permission("/v1/example/example/list")
async def get_all_examples(
limit: int = Query(50, ge=1, le=200, description="Máximo de registos por página (padrão 50, máx 200)"),
offset: int = Query(0, ge=0, description="Registos a saltar (para paginação)"),
token_data: dict = Depends(validate_token),
db: AsyncSession = Depends(get_db)
):
"""
Lista todos os exemplos com paginação.
Args:
limit: Número máximo de registos a retornar
offset: Número de registos a saltar
token_data: Dados do token de autenticação
db: Sessão da base de dados
Returns:
UniversalResponse com lista de exemplos
"""
results = await ExampleService.get_all_examples(db, limit, offset)
return UniversalResponse.success_response(
payload=[ExampleResponse.model_validate(item) for item in results]
)
@router.put("/{exampleId}", response_model=UniversalResponse, summary="Atualizar Exemplo")
@require_permission("/v1/example/example/update")
async def update_example(
exampleId: int,
example_data: ExampleRequest,
token_data: dict = Depends(validate_token),
db: AsyncSession = Depends(get_db),
log_audit_info: dict = Depends(get_log_audit_info)
):
"""
Atualiza um exemplo existente.
Args:
exampleId: ID do exemplo a ser atualizado
example_data: Novos dados do exemplo
token_data: Dados do token de autenticação
db: Sessão da base de dados
log_audit_info: Informações de auditoria
Returns:
UniversalResponse com o exemplo atualizado
"""
result = await ExampleService.update_example(db, exampleId, example_data, log_audit_info)
if not result:
return UniversalResponse.error_response(
code=404,
cause="Not Found",
message="Exemplo não encontrado"
)
return UniversalResponse.success_response(payload=ExampleResponse.model_validate(result))
@router.delete("/{exampleId}", response_model=UniversalResponse, summary="Deletar Exemplo")
@require_permission("/v1/example/example/delete")
async def delete_example(
exampleId: int,
token_data: dict = Depends(validate_token),
db: AsyncSession = Depends(get_db)
):
"""
Elimina (soft delete) um exemplo.
Args:
exampleId: ID do exemplo a ser eliminado
token_data: Dados do token de autenticação
db: Sessão da base de dados
Returns:
UniversalResponse com o exemplo eliminado
"""
result = await ExampleService.delete_example(db, exampleId)
if not result:
return UniversalResponse.error_response(
code=404,
cause="Not Found",
message="Exemplo não encontrado"
)
return UniversalResponse.success_response(payload={
"id": result.id,
"message": "Exemplo eliminado com sucesso"
})
Pontos-Chave:
exampleId)get_example)log_audit_info apenas para operações de criação/atualização/exclusão@require_permission com o caminho da permissão apropriadoPara criar sua conta de utilizador no sistema:
Configure a variável de ambiente no ficheiro .env:
API_UNICV=<url_base_da_api_sii>
Faça login na aplicação. Durante o primeiro acesso, o seu utilizador será criado automaticamente com o ID correspondente ao seu código UNICV.
Para obter permissões administrativas e gerir rotas e permissões, associe-se ao grupo Admin (ID 1) através de uma das seguintes opções:
/v1/basic/assign/user no controlador de associação de utilizadores@require_permission("/v1/basic/assign/user")@require_permission para restaurar a proteçãosiga_user_groups:INSERT INTO siga_user_groups (siga_user_id, group_id, status, is_active) VALUES (<seu_id>, 1, 'ACTIVE', true);
Após estar associado ao grupo Admin:
@require_permission("/v1/basic/routes/create") do endpoint de criação de rotas/v1/basic/routes/create/v1/basic/routes/create)/v1/basic/assign/user)Nota: Caso esta a seguir o exemplo acima deverá criar as rotas/permissoes para os endpoints(exemplo /v1/example/example/create)
Crie app/modules/example/config/routes.py:
from fastapi import APIRouter
from ..controllers.example_controller import router as example_router
exampleRoutes = APIRouter()
exampleRoutes.include_router(example_router, prefix="/v1/example", tags=["Example"])
Crie app/modules/example/config/alembic.py:
from ..models.example_model import Example
Importante: Este ficheiro garante que o Alembic pode detetar os seus modelos para migrações.
Edite main.py para incluir as rotas do seu novo módulo:
# ...código existente...
from app.modules.example.config.routes import exampleRoutes
app = FastAPI()
# ...código existente...
# Registar rotas
app.include_router(exampleRoutes)
# ...código existente...
Edite alembic/env.py para importar a configuração Alembic do seu módulo:
# ...imports existentes...
from app.modules.example.config.alembic import *
Execute os seguintes comandos para criar e aplicar a migração da base de dados:
# Gerar a migração automaticamente
alembic revision --autogenerate -m "Create example table"
Importante: Após gerar a migração, é necessário revisar o ficheiro gerado em alembic/versions/<timestamp>_create_example_table.py.
O Alembic pode tentar criar o tipo ENUM siga_status_enum que já existe no sistema. Localize a linha:
sa.Column('status', sa.Enum('ACTIVE', 'INACTIVE', name='siga_status_enum'), nullable=True),
E substitua-a por:
sa.Column('status', sa.Enum('ACTIVE', 'INACTIVE', name='siga_status_enum', create_type=False, native_enum=False), nullable=True),
Explicação: Os parâmetros create_type=False e native_enum=False indicam ao SQLAlchemy que o tipo ENUM já existe, evitando erros de duplicação.
Após fazer a correção, execute:
alembic upgrade head
Se a migração for bem-sucedida, o esquema da sua tabela será criado no PostgreSQL.
Certifique-se de que a aplicação está em execução:
uvicorn main:app --reload
Navegue até http://localhost:8000/docs
Procure a secção Example no Swagger UI
Teste as suas operações CRUD:
/v1/example/ - Criar um novo exemplo/v1/example/ - Listar todos os exemplos/v1/example/{exampleId} - Obter um exemplo específico/v1/example/{exampleId} - Atualizar um exemplo/v1/example/{exampleId} - Eliminar um exemploLogmain.pyalembic/env.py atualizado@cache_response para operações de leitura