Development challenge
from datetime import timedelta, datetime

from decouple import config
from fastapi import APIRouter, Depends, HTTPException, status
from jose import jwt
from sqlalchemy.orm import Session

from app.authentication.jwt_schema import Token
from app.database.db import SessionLocal
from app.user.user_model import User
from app.user.user_schema import UserLogin

router = APIRouter()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=int(config("TOKEN_EXPIRES")))
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, config("SECRET"), algorithm=config("ALGORITHM"))
    return encoded_jwt


@router.post("/auth/", response_model=Token)
def login_for_access_token(form_data: UserLogin, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.userid == form_data.userid).first()
    if not user or not user.verify_password(form_data.password):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password")

    access_token = create_access_token(data={"sub": user.userid})
    return {"access_token": access_token, "token_type": "bearer"}
app/authentication/auth_routes.py
micheltlutz/dev-challenge
from pydantic import BaseModel


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    userid: str = None

app/authentication/jwt_schema.py
micheltlutz/dev-challenge
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

from app.balance.balance_schema import Balance
from app.database.db import SessionLocal
from app.dependencies.current_user import get_current_user
from app.statement.statement_model import Statement

router = APIRouter()

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@router.get("/balance/", response_model=Balance, description="""
    Esta rota retorna o saldo calculado de todas as transações de um usuário.
    
    - Regra para **Depósito**: Se o usuário depositar from_user == to_user, o valor é somado ao saldo.
    - Regra para **Depósito**: Se o usuário depositar from_user != to_user, o valor é somado ao saldo.
    - Regra para **Saque**: Se o usuário sacar da sua própria conta, o valor é subtraído do saldo.
    - Regra para **Transferência**: Se o usuário transferir para sua própria conta (to_user == from_user), o valor é somado ao saldo.
    - Regra para **Transferência**: Se o usuário transferir para outra conta (to_user != from_user), o valor é subtraído do saldo.
    """)
def get_calculated_balance(db: Session = Depends(get_db), _: str = Depends(get_current_user)):
    statements = db.query(Statement).all()

    amount = 0

    for stmt in statements:
        if (stmt.type == "Deposit" and stmt.to_user == stmt.from_user
                or stmt.type == "Deposit" and stmt.to_user != stmt.from_user):
            amount += float(stmt.amount)
        elif stmt.type == "Withdrawal":
            amount -= float(stmt.amount)
        elif stmt.type == "Transfer":
            if stmt.to_user == stmt.from_user:
                amount += float(stmt.amount)
            elif stmt.from_user != stmt.to_user:
                amount -= float(stmt.amount)

    return Balance(amount=amount)
app/balance/balance_routes.py
micheltlutz/dev-challenge
from pydantic import BaseModel


class Balance(BaseModel):
    amount: float


app/balance/balance_schema.py
micheltlutz/dev-challenge
from fastapi import APIRouter, status
from fastapi.responses import JSONResponse

from app.contact.contact_schema import Contact

router = APIRouter()


@router.post("/contact/")
async def create_contact(contact: Contact):
    message = f"""Thank {contact.name}, for getting in touch and sharing your interests,
    look forward to hearing from you soon."""

    return JSONResponse(
        content={'msg': message},
        status_code=status.HTTP_200_OK
    )
app/contact/contact_routes.py
micheltlutz/dev-challenge
from pydantic import BaseModel, EmailStr


class Contact(BaseModel):
    name: str
    email: EmailStr
    message: str
    interest: str

app/contact/contact_schema.py
micheltlutz/dev-challenge
from decouple import config
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

engine = create_engine(config("DATABASE_URL"), echo=True)
with engine.connect() as conn:
    pass

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()


app/database/db.py
micheltlutz/dev-challenge
import jwt
from decouple import config
from fastapi import HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jwt import PyJWTError

security = HTTPBearer()

credentials_exception = HTTPException(
    status_code=401,
    detail="Could not validate credentials",
    headers={"WWW-Authenticate": "Bearer"},
)


def get_current_user(authorization: HTTPAuthorizationCredentials = Security(security)):
    token = authorization.credentials
    try:
        payload = jwt.decode(token, config("SECRET"), algorithms=[config("algorithm")])
        user_id = payload.get("sub")

        if user_id is None:
            raise credentials_exception
        return user_id
    except PyJWTError:
        raise credentials_exception

app/dependencies/current_user.py
micheltlutz/dev-challenge
from decouple import config
from fastapi import APIRouter, status
from fastapi.responses import JSONResponse
from fastapi.responses import RedirectResponse

router = APIRouter()


@router.get("/")
def redirect_to_docs():
    response = RedirectResponse(url='/docs')
    return response


@router.get("/health-check")
def health_check():
    return JSONResponse(
        content={"status": "ok"},
        status_code=status.HTTP_200_OK
    )


@router.get("/version")
def health_check():
    return JSONResponse(
        content={"version": config("VERSION")},
        status_code=status.HTTP_200_OK
    )

app/home/home_routes.py
micheltlutz/dev-challenge
import logging

from decouple import config
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

# Importing routes
from app.authentication import auth_routes
from app.balance import balance_routes
from app.contact import contact_routes
from app.database.db import engine, Base
from app.home import home_routes
from app.statement import statement_routes
from app.user import user_routes

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

print(config("PROJECT_NAME"))


# title=config("PROJECT_NAME")
# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Allows any origin. Don't do this in production!
    allow_credentials=True,
    allow_methods=["*"],  # Allows all methods
    allow_headers=["*"],
)


def init_db():
    try:
        Base.metadata.create_all(bind=engine)
        logging.info("Database initialized successfully.")
    except Exception as e:
        logging.error(f"Error initializing the database: {e}")


@app.on_event("startup")
async def startup_event():
    init_db()

app.include_router(auth_routes.router, tags=["authentication"])
app.include_router(home_routes.router, tags=["home"])
app.include_router(user_routes.router, tags=["users"])
app.include_router(statement_routes.router, tags=["statement"])
app.include_router(balance_routes.router, tags=["balance"])
app.include_router(contact_routes.router, tags=["contact"])

app/main.py
micheltlutz/dev-challenge
from sqlalchemy import Column, Integer, String, DateTime

from app.database.db import Base


class Statement(Base):
    __tablename__ = "statement"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    description = Column(String, unique=False)
    type = Column(String, nullable=False)
    created_at = Column(DateTime, nullable=False)
    amount = Column(String, nullable=False)
    to_user = Column(String, nullable=False)
    from_user = Column(String, nullable=False)
    bank_name = Column(String, nullable=False)
    authentication = Column(String, nullable=False)

app/statement/statement_model.py
micheltlutz/dev-challenge
import random
from datetime import datetime

from faker import Faker
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session

from app.database.db import SessionLocal
from app.dependencies.current_user import get_current_user
from app.statement.statement_model import Statement
from app.statement.statement_schema import StatementCreate

router = APIRouter()
fake = Faker()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# Create a new statement
@router.post("/statement/", status_code=201)
def create_statement(statement: StatementCreate, db: Session = Depends(get_db), _: str = Depends(get_current_user)):
    # Criar o registro de statement
    db_statement = Statement(
        description=statement.description,
        type=statement.type,
        created_at=datetime.now(),
        amount=statement.amount,
        to_user=statement.to_user,
        from_user=statement.from_user,
        bank_name=statement.bank_name,
        authentication=statement.authentication
    )

    # Adicionar ao banco de dados
    db.add(db_statement)
    db.commit()
    db.refresh(db_statement)

    return JSONResponse(
        content={'msg': 'success'},
        status_code=status.HTTP_201_CREATED
    )


@router.get("/generate-random-statement/{registers_to_generate}")
def generate_random_statements(registers_to_generate: int = 1, db: Session = Depends(get_db), _: str = Depends(get_current_user)):
    if registers_to_generate < 1 | registers_to_generate > 100:
        raise HTTPException(status_code=400, detail="Count must be greater than 0 an minor 100.")

    statements = []

    for _ in range(registers_to_generate):
        # Gerar dados aleatórios para o statement
        description = fake.sentence()
        type = random.choice(["Deposit", "Withdrawal", "Transfer"])
        amount = str(round(fake.random_number(4, True) * 0.01, 2))  # Gera um valor aleatório como 1234.56
        to_user = fake.name()
        from_user = fake.name()
        bank_name = fake.company()
        authentication = fake.sha256()

        # Criar o registro de statement
        db_statement = Statement(
            description=description,
            type=type,
            created_at=fake.date_this_decade(),
            amount=amount,
            to_user=to_user,
            from_user=from_user,
            bank_name=bank_name,
            authentication=authentication
        )

        # Adicionar ao banco de dados
        db.add(db_statement)
        statements.append(db_statement)

    db.commit()

    for stmt in statements:
        db.refresh(stmt)

    return {"message": f"{registers_to_generate} random statements generated and added.", "statements": statements}


@router.get("/statements/")
def list_statements(skip: int = 0, limit: int = 10, db: Session = Depends(get_db), _: str = Depends(get_current_user)):
    # Query the database
    statements = db.query(Statement).offset(skip).limit(limit).all()
    return statements
app/statement/statement_routes.py
micheltlutz/dev-challenge
from datetime import datetime

from pydantic import BaseModel


class StatementCreate(BaseModel):
    id: int
    description: str
    type: str
    created_at: datetime
    amount: str
    to_user: str
    from_user: str
    bank_name: str
    authentication: str

    class Config:
        from_attributes = True

app/statement/statement_schema.py
micheltlutz/dev-challenge
from passlib.context import CryptContext
from sqlalchemy import Column, Integer, String, Date

from app.database.db import Base

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    userid = Column(String, unique=True, index=True)
    fullname = Column(String, nullable=False)
    birthdate = Column(Date, nullable=False)
    hashed_password = Column(String)

    def verify_password(self, password):
        return pwd_context.verify(password, self.hashed_password)

    def hash_password(self, password):
        self.hashed_password = pwd_context.hash(password)

app/user/user_model.py
micheltlutz/dev-challenge
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session

from app.database.db import SessionLocal
from app.dependencies.current_user import get_current_user
from app.user.user_model import User
from app.user.user_schema import UserCreate, UserEdit

router = APIRouter()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# Create a new user
@router.post("/users/", status_code=201)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.userid == user.userid).first()
    if db_user:
        raise HTTPException(status_code=400, detail="Username already registered")

    new_user = User(userid=user.userid)
    new_user.hash_password(user.password)
    new_user.fullname = user.fullname
    new_user.birthdate = user.birthdate

    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return JSONResponse(
        content={'msg': 'success'},
        status_code=status.HTTP_201_CREATED
    )


@router.put("/users/{user_id}", response_model=UserEdit)
def update_user(user_id: int, user_edit: UserEdit, db: Session = Depends(get_db), _: str = Depends(get_current_user)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")

    if user_edit.password:
        db_user.hash_password(user_edit.password)
    if user_edit.fullname:
        db_user.fullname = user_edit.fullname
    if user_edit.birthdate:
        db_user.birthdate = user_edit.birthdate

    db.commit()
    db.refresh(db_user)
    return db_user

app/user/user_routes.py
micheltlutz/dev-challenge
from datetime import date

from pydantic import BaseModel, EmailStr


class UserCreate(BaseModel):
    userid: EmailStr
    password: str
    fullname: str
    birthdate: date

    class Config:
        from_attributes = True


class UserEdit(BaseModel):
    password: str = None
    fullname: str = None
    birthdate: date = None


class User(BaseModel):
    id: int
    userid: EmailStr
    fullname: str
    birthdate: date


class UserLogin(BaseModel):
    userid: str
    password: str

app/user/user_schema.py
micheltlutz/dev-challenge
from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.home import home_routes

appT = FastAPI()


client = TestClient(appT)

appT.include_router(home_routes.router)


def test_health_check():
    response = client.get("/health-check")
    assert response.status_code == 200
    assert response.json() == {"status": "ok"}


def test_version():
    response = client.get("/version")
    assert response.status_code == 200
    assert response.json() == {"version": "1.0.0"}
    
tests/test_home_routes.py
micheltlutz/dev-challenge