initial commit 2

backend/app/auth.py

@@ -0,0 +1,22 @@
+from datetime import datetime, timedelta, timezone
+
+from jose import jwt
+from passlib.context import CryptContext
+
+from .core.config import settings
+
+pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')
+
+
+def hash_password(password: str) -> str:
+    return pwd_context.hash(password)
+
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+def create_access_token(*, user_id: int, email: str) -> str:
+    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes)
+    payload = {'sub': str(user_id), 'email': email, 'exp': expire}
+    return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm)

backend/app/core/config.py

@@ -0,0 +1,25 @@
+from functools import lru_cache
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file='.env', env_file_encoding='utf-8', extra='ignore')
+
+    database_url: str = 'postgresql+psycopg://postgres:password@127.0.0.1:5432/hackaton'
+    secret_key: str = 'change-this-in-production'
+    algorithm: str = 'HS256'
+    access_token_expire_minutes: int = 60 * 24 * 7
+    cors_origins: str = 'http://localhost:3000,http://10.0.2.2:3000'
+
+    @property
+    def cors_origin_list(self) -> list[str]:
+        return [origin.strip() for origin in self.cors_origins.split(',') if origin.strip()]
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()
+
+
+settings = get_settings()

backend/app/crud.py

@@ -0,0 +1,56 @@
+from datetime import datetime, timezone
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from .auth import hash_password, verify_password
+from .models import Address, User
+from .schemas import AddressCreate, UserCreate
+
+
+def get_user_by_id(db: Session, user_id: int) -> User | None:
+    return db.get(User, user_id)
+
+
+def get_user_by_email(db: Session, email: str) -> User | None:
+    statement = select(User).where(User.email == email)
+    return db.scalar(statement)
+
+
+def create_user(db: Session, user_in: UserCreate) -> User:
+    user = User(
+        name=user_in.name,
+        email=user_in.email,
+        password_hash=hash_password(user_in.password),
+    )
+    db.add(user)
+    db.commit()
+    db.refresh(user)
+    return user
+
+
+def authenticate_user(db: Session, email: str, password: str) -> User | None:
+    user = get_user_by_email(db, email)
+    if user is None:
+        return None
+    if not verify_password(password, user.password_hash):
+        return None
+
+    user.last_login_at = datetime.now(timezone.utc)
+    db.add(user)
+    db.commit()
+    db.refresh(user)
+    return user
+
+
+def create_address(db: Session, user: User, address_in: AddressCreate) -> Address:
+    address = Address(
+        user_id=user.id,
+        house_number=address_in.house_number,
+        colonia=address_in.colonia,
+        street=address_in.street,
+    )
+    db.add(address)
+    db.commit()
+    db.refresh(address)
+    return address

backend/app/db/session.py

@@ -0,0 +1,22 @@
+from collections.abc import Generator
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
+
+from ..core.config import settings
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+engine = create_engine(settings.database_url, pool_pre_ping=True)
+SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
+
+
+def get_db() -> Generator[Session, None, None]:
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

backend/app/dependencies.py

@@ -0,0 +1,31 @@
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from jose import JWTError, jwt
+from sqlalchemy.orm import Session
+
+from .core.config import settings
+from .crud import get_user_by_id
+from .db.session import get_db
+from .models import User
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/auth/login')
+
+
+def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)) -> User:
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail='Token inválido o expirado',
+        headers={'WWW-Authenticate': 'Bearer'},
+    )
+    try:
+        payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
+        user_id = payload.get('sub')
+        if user_id is None:
+            raise credentials_exception
+    except JWTError as exc:
+        raise credentials_exception from exc
+
+    user = get_user_by_id(db, int(user_id))
+    if user is None:
+        raise credentials_exception
+    return user

backend/app/main.py

@@ -0,0 +1,79 @@
+from fastapi import Depends, FastAPI, HTTPException, status
+from fastapi.middleware.cors import CORSMiddleware
+from sqlalchemy.orm import Session
+
+from .auth import create_access_token
+from .core.config import settings
+from .crud import authenticate_user, create_address, create_user, get_user_by_email
+from .db.session import Base, engine, get_db
+from .dependencies import get_current_user
+from .models import Address, User
+from .schemas import AddressCreate, AddressRead, TokenResponse, UserCreate, UserLogin, UserRead
+
+app = FastAPI(title='Flutter Auth API', version='1.0.0')
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=settings.cors_origin_list,
+    allow_credentials=True,
+    allow_methods=['*'],
+    allow_headers=['*'],
+)
+
+
+@app.on_event('startup')
+def on_startup() -> None:
+    Base.metadata.create_all(bind=engine)
+
+
+@app.get('/health')
+def health_check() -> dict[str, str]:
+    return {'status': 'ok'}
+
+
+@app.post('/auth/register', response_model=TokenResponse, status_code=status.HTTP_201_CREATED)
+def register(payload: UserCreate, db: Session = Depends(get_db)) -> TokenResponse:
+    existing_user = get_user_by_email(db, payload.email)
+    if existing_user is not None:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='Ya existe un usuario con ese correo')
+
+    user = create_user(db, payload)
+    token = create_access_token(user_id=user.id, email=user.email)
+    return TokenResponse(token=token, user=UserRead.model_validate(user))
+
+
+@app.post('/auth/login', response_model=TokenResponse)
+def login(payload: UserLogin, db: Session = Depends(get_db)) -> TokenResponse:
+    user = authenticate_user(db, payload.email, payload.password)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Correo o contraseña inválidos')
+
+    token = create_access_token(user_id=user.id, email=user.email)
+    return TokenResponse(token=token, user=UserRead.model_validate(user))
+
+
+@app.get('/me', response_model=UserRead)
+def read_me(current_user: User = Depends(get_current_user)) -> UserRead:
+    return UserRead.model_validate(current_user)
+
+
+@app.post('/addresses', response_model=AddressRead, status_code=status.HTTP_201_CREATED)
+def save_address(
+    payload: AddressCreate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+) -> AddressRead:
+    if payload.email is not None and payload.email != current_user.email:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='El correo no coincide con el usuario autenticado')
+
+    address = create_address(db, current_user, payload)
+    return AddressRead.model_validate(address)
+
+
+@app.get('/addresses/me', response_model=list[AddressRead])
+def read_my_addresses(
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+) -> list[AddressRead]:
+    addresses = sorted(current_user.addresses, key=lambda address: address.created_at, reverse=True)
+    return [AddressRead.model_validate(address) for address in addresses]

backend/app/models.py

@@ -0,0 +1,32 @@
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from .db.session import Base
+
+
+class User(Base):
+    __tablename__ = 'users'
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    name: Mapped[str] = mapped_column(String(120), nullable=False)
+    email: Mapped[str] = mapped_column(String(255), unique=True, index=True, nullable=False)
+    password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
+    last_login_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), nullable=False)
+
+    addresses: Mapped[list['Address']] = relationship(back_populates='user', cascade='all, delete-orphan')
+
+
+class Address(Base):
+    __tablename__ = 'addresses'
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey('users.id', ondelete='CASCADE'), index=True, nullable=False)
+    house_number: Mapped[str] = mapped_column(String(50), nullable=False)
+    colonia: Mapped[str] = mapped_column(String(120), nullable=False)
+    street: Mapped[str] = mapped_column(String(160), nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), nullable=False)
+
+    user: Mapped['User'] = relationship(back_populates='addresses')

backend/app/schemas.py

@@ -0,0 +1,50 @@
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict, EmailStr, Field
+
+
+class UserBase(BaseModel):
+    name: str
+    email: EmailStr
+
+
+class UserCreate(UserBase):
+    password: str
+
+
+class UserLogin(BaseModel):
+    email: EmailStr
+    password: str
+
+
+class UserRead(UserBase):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    last_login_at: datetime | None = None
+    created_at: datetime
+
+
+class TokenResponse(BaseModel):
+    token: str
+    user: UserRead
+
+
+class AddressCreate(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    house_number: str = Field(alias='houseNumber')
+    colonia: str
+    street: str
+    email: EmailStr | None = None
+
+
+class AddressRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    house_number: str
+    colonia: str
+    street: str
+    user_id: int
+    created_at: datetime