import uuid import logging import os import smtplib from datetime import datetime, timedelta, timezone from email.message import EmailMessage from email_validator import EmailNotValidError, validate_email from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request, status from pydantic import BaseModel, Field from sqlalchemy import func, or_, select from sqlalchemy.orm import Session as DbSession from auth import ( create_access_token, create_refresh_token, create_one_time_token, decode_access_token, hash_password, hash_refresh_token, hash_one_time_token, normalize_email, refresh_expiry, verify_password, ) from db import check_database, engine from models import AuditEvent, AuthRateLimit, EmailVerificationToken, Entitlement, PasswordResetToken, Ride, Session, Subscription, TelemetryPoint, User, Vehicle logger = logging.getLogger(__name__) app = FastAPI( title="MotorSignal API", version="1.0.0", docs_url=None, redoc_url=None, openapi_url=None, ) class RegisterRequest(BaseModel): email: str password: str = Field(min_length=12, max_length=128) device_name: str | None = Field(default=None, max_length=255) class LoginRequest(BaseModel): email: str password: str device_name: str | None = Field(default=None, max_length=255) class RefreshRequest(BaseModel): refresh_token: str class EmailRequest(BaseModel): email: str class TokenRequest(BaseModel): token: str = Field(min_length=20, max_length=256) class PasswordResetRequest(TokenRequest): password: str = Field(min_length=12, max_length=128) class TokenResponse(BaseModel): access_token: str refresh_token: str token_type: str = "bearer" class SessionResponse(BaseModel): id: uuid.UUID device_name: str | None created_at: datetime expires_at: datetime last_used_at: datetime | None current: bool class AdminUserCreate(BaseModel): email: str password: str = Field(min_length=12, max_length=128) role: str = "user" class AdminUserUpdate(BaseModel): email: str | None = None role: str | None = None is_active: bool | None = None email_verified: bool | None = None class VehicleCreate(BaseModel): user_id: uuid.UUID nickname: str = Field(min_length=1, max_length=120) make: str = Field(min_length=1, max_length=120) model: str = Field(min_length=1, max_length=120) year: int | None = Field(default=None, ge=1885, le=2100) vin: str | None = Field(default=None, min_length=11, max_length=17) class VehicleUpdate(BaseModel): nickname: str | None = Field(default=None, min_length=1, max_length=120) make: str | None = Field(default=None, min_length=1, max_length=120) model: str | None = Field(default=None, min_length=1, max_length=120) year: int | None = Field(default=None, ge=1885, le=2100) vin: str | None = Field(default=None, min_length=11, max_length=17) class RideCreate(BaseModel): user_id: uuid.UUID vehicle_id: uuid.UUID | None = None started_at: datetime ended_at: datetime | None = None distance_miles: float | None = Field(default=None, ge=0) top_speed_mph: float | None = Field(default=None, ge=0) avg_speed_mph: float | None = Field(default=None, ge=0) status: str = Field(default="completed", min_length=1, max_length=32) class RideUpdate(BaseModel): vehicle_id: uuid.UUID | None = None ended_at: datetime | None = None distance_miles: float | None = Field(default=None, ge=0) top_speed_mph: float | None = Field(default=None, ge=0) avg_speed_mph: float | None = Field(default=None, ge=0) status: str | None = Field(default=None, min_length=1, max_length=32) class UserVehicleCreate(BaseModel): nickname: str = Field(min_length=1, max_length=120) make: str = Field(min_length=1, max_length=120) model: str = Field(min_length=1, max_length=120) year: int | None = Field(default=None, ge=1885, le=2100) vin: str | None = Field(default=None, min_length=11, max_length=17) class UserRideCreate(BaseModel): vehicle_id: uuid.UUID | None = None started_at: datetime ended_at: datetime | None = None distance_miles: float | None = Field(default=None, ge=0) top_speed_mph: float | None = Field(default=None, ge=0) avg_speed_mph: float | None = Field(default=None, ge=0) status: str = Field(default="completed", min_length=1, max_length=32) class TelemetryPointInput(BaseModel): recorded_at: datetime latitude: float | None = Field(default=None, ge=-90, le=90) longitude: float | None = Field(default=None, ge=-180, le=180) speed_mph: float | None = Field(default=None, ge=0) lean_angle_deg: float | None = Field(default=None, ge=-90, le=90) g_force: float | None = Field(default=None, ge=0) accel_g: float | None = Field(default=None, ge=-10, le=10) brake_g: float | None = Field(default=None, ge=-10, le=10) class TelemetryBatch(BaseModel): points: list[TelemetryPointInput] = Field(min_length=1, max_length=1000) class SubscriptionUpdate(BaseModel): plan: str = Field(min_length=1, max_length=32) status: str = Field(min_length=1, max_length=32) current_period_start: datetime | None = None current_period_end: datetime | None = None class EntitlementCreate(BaseModel): key: str = Field(min_length=1, max_length=96) value: dict | None = None expires_at: datetime | None = None def get_db(): db = DbSession(engine) try: yield db finally: db.close() def enforce_auth_rate_limit(request: Request, db: DbSession, action: str, limit: int = 10, window_seconds: int = 900) -> None: client = request.client.host if request.client else "unknown" key = f"{action}:{client}" now = datetime.now(timezone.utc) bucket = db.get(AuthRateLimit, key, with_for_update=True) if bucket is None: db.add(AuthRateLimit(key=key, window_started=now, attempts=1)) db.commit() return elapsed = (now - bucket.window_started).total_seconds() if elapsed >= window_seconds: bucket.window_started = now bucket.attempts = 1 elif bucket.attempts >= limit: retry_after = max(1, int(window_seconds - elapsed)) raise HTTPException(status_code=429, detail="Too many authentication attempts", headers={"Retry-After": str(retry_after)}) else: bucket.attempts += 1 db.commit() def extract_bearer_token(authorization: str | None) -> str: if not authorization: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing authorization token", ) scheme, _, token = authorization.partition(" ") if scheme.lower() != "bearer" or not token: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authorization header", ) return token def get_request_context(request: Request) -> tuple[str | None, str | None]: return request.client.host if request.client else None, request.headers.get("user-agent") def audit( db: DbSession, event_type: str, actor_user_id: uuid.UUID | None = None, target_user_id: uuid.UUID | None = None, request: Request | None = None, metadata: dict | None = None, ) -> None: ip_address = request.client.host if request and request.client else None user_agent = request.headers.get("user-agent")[:512] if request and request.headers.get("user-agent") else None db.add(AuditEvent(event_type=event_type, actor_user_id=actor_user_id, target_user_id=target_user_id, ip_address=ip_address, user_agent=user_agent, metadata_json=metadata)) def send_email(recipient: str, subject: str, body: str) -> None: host = os.getenv("SMTP_HOST") if not host: return try: message = EmailMessage() message["From"] = os.environ["SMTP_FROM"] message["To"] = recipient message["Subject"] = subject message.set_content(body) port = int(os.getenv("SMTP_PORT", "587")) with smtplib.SMTP(host, port, timeout=10) as smtp: if os.getenv("SMTP_STARTTLS", "true").lower() == "true": smtp.starttls() if os.getenv("SMTP_USER"): smtp.login(os.environ["SMTP_USER"], os.environ["SMTP_PASSWORD"]) smtp.send_message(message) except (KeyError, OSError, smtplib.SMTPException): logger.exception("MotorSignal email delivery failed") def get_current_identity( authorization: str | None = Header(default=None), db: DbSession = Depends(get_db), ) -> tuple[User, Session]: token = extract_bearer_token(authorization) try: payload = decode_access_token(token) except Exception: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired access token", ) if payload.get("type") != "access": raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid access token", ) try: user_id = uuid.UUID(payload["sub"]) session_id = uuid.UUID(payload["sid"]) except Exception: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid access token", ) session = db.get(Session, session_id) if ( session is None or session.user_id != user_id or session.revoked_at is not None or session.expires_at <= datetime.now(timezone.utc) ): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Session is no longer valid", ) user = db.get(User, user_id) if user is None or not user.is_active: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="User is unavailable", ) return user, session def get_current_user(identity: tuple[User, Session] = Depends(get_current_identity)) -> User: return identity[0] def require_roles(*roles: str): def dependency(user: User = Depends(get_current_user)) -> User: if user.role not in roles: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions") return user return dependency @app.get("/api/v1/health") def health(): return { "status": "ok", "service": "motorsignal-api", } @app.get("/api/v1/ready") def readiness(): try: check_database() return { "status": "ready", "database": "ok", } except Exception: raise HTTPException( status_code=503, detail="Service unavailable", ) @app.post( "/api/v1/auth/register", response_model=TokenResponse, status_code=201, ) def register( request: RegisterRequest, http_request: Request, db: DbSession = Depends(get_db), ): enforce_auth_rate_limit(http_request, db, "register") try: validated = validate_email( request.email, check_deliverability=False, ) email = normalize_email(validated.normalized) except EmailNotValidError: raise HTTPException( status_code=400, detail="Invalid email address", ) existing = db.scalar( select(User).where(User.email == email) ) if existing is not None: raise HTTPException( status_code=409, detail="Account already exists", ) user = User( email=email, password_hash=hash_password(request.password), ) db.add(user) db.flush() raw_refresh, refresh_hash = create_refresh_token() session = Session( user_id=user.id, refresh_token_hash=refresh_hash, device_name=request.device_name, expires_at=refresh_expiry(), ) db.add(session) verification_raw, verification_hash = create_one_time_token() db.add(EmailVerificationToken(user_id=user.id, token_hash=verification_hash, expires_at=datetime.now(timezone.utc) + timedelta(hours=24))) audit(db, "auth.register", actor_user_id=user.id, target_user_id=user.id) db.commit() web_base = os.getenv("WEB_BASE_URL", "https://motorsignal.app") send_email(email, "Verify your MotorSignal email", f"Verify your email: {web_base}/verify-email?token={verification_raw}") return TokenResponse( access_token=create_access_token(user.id, session.id), refresh_token=raw_refresh, ) @app.post( "/api/v1/auth/login", response_model=TokenResponse, ) def login( request: LoginRequest, http_request: Request, db: DbSession = Depends(get_db), ): enforce_auth_rate_limit(http_request, db, "login") email = normalize_email(request.email) user = db.scalar( select(User).where(User.email == email) ) if user is None or not verify_password( request.password, user.password_hash, ): raise HTTPException( status_code=401, detail="Invalid email or password", ) if not user.is_active: raise HTTPException( status_code=403, detail="Account disabled", ) raw_refresh, refresh_hash = create_refresh_token() session = Session( user_id=user.id, refresh_token_hash=refresh_hash, device_name=request.device_name, expires_at=refresh_expiry(), ) db.add(session) audit(db, "auth.login", actor_user_id=user.id, target_user_id=user.id) db.commit() return TokenResponse( access_token=create_access_token(user.id, session.id), refresh_token=raw_refresh, ) @app.post( "/api/v1/auth/refresh", response_model=TokenResponse, ) def refresh( request: RefreshRequest, db: DbSession = Depends(get_db), ): old_hash = hash_refresh_token(request.refresh_token) session = db.scalar( select(Session).where( Session.refresh_token_hash == old_hash ) ) now = datetime.now(timezone.utc) if ( session is None or session.revoked_at is not None or session.expires_at <= now ): raise HTTPException( status_code=401, detail="Invalid refresh token", ) user = db.get(User, session.user_id) if user is None or not user.is_active: raise HTTPException( status_code=401, detail="Invalid refresh token", ) new_raw, new_hash = create_refresh_token() session.refresh_token_hash = new_hash session.last_used_at = now session.expires_at = refresh_expiry() db.commit() return TokenResponse( access_token=create_access_token(user.id, session.id), refresh_token=new_raw, ) @app.post("/api/v1/auth/logout", status_code=204) def logout( request: RefreshRequest, db: DbSession = Depends(get_db), ): refresh_hash = hash_refresh_token(request.refresh_token) session = db.scalar( select(Session).where( Session.refresh_token_hash == refresh_hash ) ) if session is not None and session.revoked_at is None: session.revoked_at = datetime.now(timezone.utc) audit(db, "auth.logout", actor_user_id=session.user_id, target_user_id=session.user_id) db.commit() return None @app.get("/api/v1/me") def me(user: User = Depends(get_current_user)): return { "id": str(user.id), "email": user.email, "email_verified": user.email_verified, "created_at": user.created_at, } @app.post("/api/v1/auth/verify-email/request", status_code=202) def request_email_verification(payload: EmailRequest, http_request: Request, db: DbSession = Depends(get_db)): enforce_auth_rate_limit(http_request, db, "verify-email-request") email = normalize_email(payload.email) user = db.scalar(select(User).where(User.email == email)) if user is not None and not user.email_verified and user.is_active: raw, token_hash = create_one_time_token() db.add(EmailVerificationToken(user_id=user.id, token_hash=token_hash, expires_at=datetime.now(timezone.utc) + timedelta(hours=24))) audit(db, "auth.email_verification_requested", target_user_id=user.id) db.commit() web_base = os.getenv("WEB_BASE_URL", "https://motorsignal.app") send_email(email, "Verify your MotorSignal email", f"Verify your email: {web_base}/verify-email?token={raw}") return {"detail": "If the account can be verified, instructions have been sent."} @app.post("/api/v1/auth/verify-email", status_code=204) def verify_email(request: TokenRequest, http_request: Request, db: DbSession = Depends(get_db)): enforce_auth_rate_limit(http_request, db, "verify-email") token = db.scalar(select(EmailVerificationToken).where(EmailVerificationToken.token_hash == hash_one_time_token(request.token))) now = datetime.now(timezone.utc) if token is None or token.used_at is not None or token.expires_at <= now: raise HTTPException(status_code=400, detail="Invalid or expired verification token") user = db.get(User, token.user_id) if user is None: raise HTTPException(status_code=400, detail="Invalid verification token") user.email_verified = True token.used_at = now audit(db, "auth.email_verified", actor_user_id=user.id, target_user_id=user.id) db.commit() return None @app.post("/api/v1/auth/password-reset/request", status_code=202) def request_password_reset(payload: EmailRequest, http_request: Request, db: DbSession = Depends(get_db)): enforce_auth_rate_limit(http_request, db, "password-reset-request") email = normalize_email(payload.email) user = db.scalar(select(User).where(User.email == email)) if user is not None and user.is_active: raw, token_hash = create_one_time_token() db.add(PasswordResetToken(user_id=user.id, token_hash=token_hash, expires_at=datetime.now(timezone.utc) + timedelta(hours=1))) audit(db, "auth.password_reset_requested", target_user_id=user.id) db.commit() web_base = os.getenv("WEB_BASE_URL", "https://motorsignal.app") send_email(email, "Reset your MotorSignal password", f"Reset your password: {web_base}/forgot-password?token={raw}") return {"detail": "If the account exists, reset instructions have been sent."} @app.post("/api/v1/auth/password-reset", status_code=204) def reset_password(request: PasswordResetRequest, http_request: Request, db: DbSession = Depends(get_db)): enforce_auth_rate_limit(http_request, db, "password-reset") token = db.scalar(select(PasswordResetToken).where(PasswordResetToken.token_hash == hash_one_time_token(request.token))) now = datetime.now(timezone.utc) if token is None or token.used_at is not None or token.expires_at <= now: raise HTTPException(status_code=400, detail="Invalid or expired reset token") user = db.get(User, token.user_id) if user is None or not user.is_active: raise HTTPException(status_code=400, detail="Invalid reset token") user.password_hash = hash_password(request.password) token.used_at = now db.query(Session).filter(Session.user_id == user.id, Session.revoked_at.is_(None)).update({Session.revoked_at: now}, synchronize_session=False) audit(db, "auth.password_reset_completed", actor_user_id=user.id, target_user_id=user.id) db.commit() return None @app.get("/api/v1/auth/sessions", response_model=list[SessionResponse]) def list_sessions(identity: tuple[User, Session] = Depends(get_current_identity), db: DbSession = Depends(get_db)): user, current = identity sessions = db.scalars(select(Session).where(Session.user_id == user.id, Session.revoked_at.is_(None)).order_by(Session.created_at.desc())).all() return [SessionResponse(id=item.id, device_name=item.device_name, created_at=item.created_at, expires_at=item.expires_at, last_used_at=item.last_used_at, current=item.id == current.id) for item in sessions] @app.delete("/api/v1/auth/sessions/{session_id}", status_code=204) def revoke_session(session_id: uuid.UUID, identity: tuple[User, Session] = Depends(get_current_identity), db: DbSession = Depends(get_db)): user, _ = identity session = db.scalar(select(Session).where(Session.id == session_id, Session.user_id == user.id)) if session is None: raise HTTPException(status_code=404, detail="Session not found") if session.revoked_at is None: session.revoked_at = datetime.now(timezone.utc) audit(db, "auth.session_revoked", actor_user_id=user.id, target_user_id=user.id, metadata={"session_id": str(session.id)}) db.commit() return None @app.post("/api/v1/auth/logout-all", status_code=204) def logout_all(identity: tuple[User, Session] = Depends(get_current_identity), db: DbSession = Depends(get_db)): user, _ = identity now = datetime.now(timezone.utc) db.query(Session).filter(Session.user_id == user.id, Session.revoked_at.is_(None)).update({Session.revoked_at: now}, synchronize_session=False) audit(db, "auth.logout_all", actor_user_id=user.id, target_user_id=user.id) db.commit() return None @app.get("/api/v1/admin/audit-events") def list_audit_events(user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): events = db.scalars(select(AuditEvent).order_by(AuditEvent.created_at.desc()).limit(100)).all() return [{"id": str(event.id), "event_type": event.event_type, "actor_user_id": str(event.actor_user_id) if event.actor_user_id else None, "target_user_id": str(event.target_user_id) if event.target_user_id else None, "created_at": event.created_at, "metadata": event.metadata_json} for event in events] def user_payload(user: User) -> dict: return {"id": str(user.id), "email": user.email, "role": user.role, "is_active": user.is_active, "email_verified": user.email_verified, "created_at": user.created_at} def vehicle_payload(vehicle: Vehicle) -> dict: return {"id": str(vehicle.id), "user_id": str(vehicle.user_id), "nickname": vehicle.nickname, "make": vehicle.make, "model": vehicle.model, "year": vehicle.year, "vin": vehicle.vin, "created_at": vehicle.created_at, "updated_at": vehicle.updated_at} def ride_payload(ride: Ride) -> dict: return {"id": str(ride.id), "user_id": str(ride.user_id), "vehicle_id": str(ride.vehicle_id) if ride.vehicle_id else None, "started_at": ride.started_at, "ended_at": ride.ended_at, "distance_miles": ride.distance_miles, "top_speed_mph": ride.top_speed_mph, "avg_speed_mph": ride.avg_speed_mph, "status": ride.status, "created_at": ride.created_at} def validate_role(role: str) -> str: if role not in {"user", "support", "admin"}: raise HTTPException(status_code=400, detail="Invalid role") return role @app.get("/api/v1/admin/users") def list_users(page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=100), search: str | None = None, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): query = select(User) count_query = select(func.count()).select_from(User) if search: pattern = f"%{search.strip().lower()}%" condition = or_(func.lower(User.email).like(pattern), User.role == search.strip().lower()) query = query.where(condition) count_query = count_query.where(condition) total = db.scalar(count_query) or 0 users = db.scalars(query.order_by(User.created_at.desc()).offset((page - 1) * page_size).limit(page_size)).all() return {"items": [user_payload(item) for item in users], "page": page, "page_size": page_size, "total": total} @app.post("/api/v1/admin/users", status_code=201) def create_user(request: AdminUserCreate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): email = normalize_email(request.email) try: email = normalize_email(validate_email(email, check_deliverability=False).normalized) except EmailNotValidError: raise HTTPException(status_code=400, detail="Invalid email address") if db.scalar(select(User).where(User.email == email)) is not None: raise HTTPException(status_code=409, detail="Account already exists") user = User(email=email, password_hash=hash_password(request.password), role=validate_role(request.role)) db.add(user) db.flush() audit(db, "admin.user_created", actor_user_id=actor.id, target_user_id=user.id) db.commit() db.refresh(user) return user_payload(user) @app.get("/api/v1/admin/users/{user_id}") def get_user(user_id: uuid.UUID, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): target = db.get(User, user_id) if target is None: raise HTTPException(status_code=404, detail="User not found") return user_payload(target) @app.patch("/api/v1/admin/users/{user_id}") def update_user(user_id: uuid.UUID, request: AdminUserUpdate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): target = db.get(User, user_id) if target is None: raise HTTPException(status_code=404, detail="User not found") changes = request.model_dump(exclude_unset=True) if "email" in changes: try: changes["email"] = normalize_email(validate_email(changes["email"], check_deliverability=False).normalized) except EmailNotValidError: raise HTTPException(status_code=400, detail="Invalid email address") duplicate = db.scalar(select(User).where(User.email == changes["email"], User.id != target.id)) if duplicate is not None: raise HTTPException(status_code=409, detail="Account already exists") if "role" in changes: changes["role"] = validate_role(changes["role"]) for field, value in changes.items(): setattr(target, field, value) audit(db, "admin.user_updated", actor_user_id=actor.id, target_user_id=target.id, metadata={"fields": list(changes)}) db.commit() db.refresh(target) return user_payload(target) @app.delete("/api/v1/admin/users/{user_id}", status_code=204) def deactivate_user(user_id: uuid.UUID, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): target = db.get(User, user_id) if target is None: raise HTTPException(status_code=404, detail="User not found") target.is_active = False now = datetime.now(timezone.utc) db.query(Session).filter(Session.user_id == target.id, Session.revoked_at.is_(None)).update({Session.revoked_at: now}, synchronize_session=False) audit(db, "admin.user_deactivated", actor_user_id=actor.id, target_user_id=target.id) db.commit() return None @app.get("/api/v1/admin/vehicles") def list_vehicles(page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=100), user_id: uuid.UUID | None = None, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): query = select(Vehicle) count_query = select(func.count()).select_from(Vehicle) if user_id: query = query.where(Vehicle.user_id == user_id) count_query = count_query.where(Vehicle.user_id == user_id) total = db.scalar(count_query) or 0 vehicles = db.scalars(query.order_by(Vehicle.created_at.desc()).offset((page - 1) * page_size).limit(page_size)).all() return {"items": [vehicle_payload(item) for item in vehicles], "page": page, "page_size": page_size, "total": total} @app.post("/api/v1/admin/vehicles", status_code=201) def create_vehicle(request: VehicleCreate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): if db.get(User, request.user_id) is None: raise HTTPException(status_code=404, detail="User not found") vehicle = Vehicle(**request.model_dump()) db.add(vehicle) db.flush() audit(db, "admin.vehicle_created", actor_user_id=actor.id, target_user_id=vehicle.user_id, metadata={"vehicle_id": str(vehicle.id)}) db.commit() db.refresh(vehicle) return vehicle_payload(vehicle) @app.get("/api/v1/admin/vehicles/{vehicle_id}") def get_vehicle(vehicle_id: uuid.UUID, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): vehicle = db.get(Vehicle, vehicle_id) if vehicle is None: raise HTTPException(status_code=404, detail="Vehicle not found") return vehicle_payload(vehicle) @app.patch("/api/v1/admin/vehicles/{vehicle_id}") def update_vehicle(vehicle_id: uuid.UUID, request: VehicleUpdate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): vehicle = db.get(Vehicle, vehicle_id) if vehicle is None: raise HTTPException(status_code=404, detail="Vehicle not found") changes = request.model_dump(exclude_unset=True) for field, value in changes.items(): setattr(vehicle, field, value) audit(db, "admin.vehicle_updated", actor_user_id=actor.id, target_user_id=vehicle.user_id, metadata={"vehicle_id": str(vehicle.id), "fields": list(changes)}) db.commit() db.refresh(vehicle) return vehicle_payload(vehicle) @app.delete("/api/v1/admin/vehicles/{vehicle_id}", status_code=204) def delete_vehicle(vehicle_id: uuid.UUID, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): vehicle = db.get(Vehicle, vehicle_id) if vehicle is None: raise HTTPException(status_code=404, detail="Vehicle not found") audit(db, "admin.vehicle_deleted", actor_user_id=actor.id, target_user_id=vehicle.user_id, metadata={"vehicle_id": str(vehicle.id)}) db.delete(vehicle) db.commit() return None @app.get("/api/v1/admin/rides") def list_rides(page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=100), user_id: uuid.UUID | None = None, vehicle_id: uuid.UUID | None = None, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): query = select(Ride) count_query = select(func.count()).select_from(Ride) if user_id: query = query.where(Ride.user_id == user_id) count_query = count_query.where(Ride.user_id == user_id) if vehicle_id: query = query.where(Ride.vehicle_id == vehicle_id) count_query = count_query.where(Ride.vehicle_id == vehicle_id) total = db.scalar(count_query) or 0 rides = db.scalars(query.order_by(Ride.started_at.desc()).offset((page - 1) * page_size).limit(page_size)).all() return {"items": [ride_payload(item) for item in rides], "page": page, "page_size": page_size, "total": total} @app.post("/api/v1/admin/rides", status_code=201) def create_ride(request: RideCreate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): if db.get(User, request.user_id) is None: raise HTTPException(status_code=404, detail="User not found") if request.vehicle_id is not None: vehicle = db.get(Vehicle, request.vehicle_id) if vehicle is None or vehicle.user_id != request.user_id: raise HTTPException(status_code=400, detail="Vehicle does not belong to user") ride = Ride(**request.model_dump()) db.add(ride) db.flush() audit(db, "admin.ride_created", actor_user_id=actor.id, target_user_id=ride.user_id, metadata={"ride_id": str(ride.id)}) db.commit() db.refresh(ride) return ride_payload(ride) @app.get("/api/v1/admin/rides/{ride_id}") def get_ride(ride_id: uuid.UUID, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): ride = db.get(Ride, ride_id) if ride is None: raise HTTPException(status_code=404, detail="Ride not found") return ride_payload(ride) @app.patch("/api/v1/admin/rides/{ride_id}") def update_ride(ride_id: uuid.UUID, request: RideUpdate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): ride = db.get(Ride, ride_id) if ride is None: raise HTTPException(status_code=404, detail="Ride not found") changes = request.model_dump(exclude_unset=True) if "vehicle_id" in changes and changes["vehicle_id"] is not None: vehicle = db.get(Vehicle, changes["vehicle_id"]) if vehicle is None or vehicle.user_id != ride.user_id: raise HTTPException(status_code=400, detail="Vehicle does not belong to user") for field, value in changes.items(): setattr(ride, field, value) audit(db, "admin.ride_updated", actor_user_id=actor.id, target_user_id=ride.user_id, metadata={"ride_id": str(ride.id), "fields": list(changes)}) db.commit() db.refresh(ride) return ride_payload(ride) @app.delete("/api/v1/admin/rides/{ride_id}", status_code=204) def archive_ride(ride_id: uuid.UUID, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): ride = db.get(Ride, ride_id) if ride is None: raise HTTPException(status_code=404, detail="Ride not found") ride.status = "archived" audit(db, "admin.ride_archived", actor_user_id=actor.id, target_user_id=ride.user_id, metadata={"ride_id": str(ride.id)}) db.commit() return None @app.get("/api/v1/vehicles") def list_my_vehicles(user: User = Depends(get_current_user), db: DbSession = Depends(get_db)): vehicles = db.scalars(select(Vehicle).where(Vehicle.user_id == user.id).order_by(Vehicle.created_at.desc())).all() return [vehicle_payload(vehicle) for vehicle in vehicles] @app.post("/api/v1/vehicles", status_code=201) def create_my_vehicle(request: UserVehicleCreate, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)): vehicle = Vehicle(user_id=user.id, **request.model_dump()) db.add(vehicle) db.flush() audit(db, "vehicle.created", actor_user_id=user.id, target_user_id=user.id, metadata={"vehicle_id": str(vehicle.id)}) db.commit() db.refresh(vehicle) return vehicle_payload(vehicle) @app.get("/api/v1/vehicles/{vehicle_id}") def get_my_vehicle(vehicle_id: uuid.UUID, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)): vehicle = db.scalar(select(Vehicle).where(Vehicle.id == vehicle_id, Vehicle.user_id == user.id)) if vehicle is None: raise HTTPException(status_code=404, detail="Vehicle not found") return vehicle_payload(vehicle) @app.patch("/api/v1/vehicles/{vehicle_id}") def update_my_vehicle(vehicle_id: uuid.UUID, request: VehicleUpdate, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)): vehicle = db.scalar(select(Vehicle).where(Vehicle.id == vehicle_id, Vehicle.user_id == user.id)) if vehicle is None: raise HTTPException(status_code=404, detail="Vehicle not found") changes = request.model_dump(exclude_unset=True) for field, value in changes.items(): setattr(vehicle, field, value) audit(db, "vehicle.updated", actor_user_id=user.id, target_user_id=user.id, metadata={"vehicle_id": str(vehicle.id), "fields": list(changes)}) db.commit() db.refresh(vehicle) return vehicle_payload(vehicle) @app.delete("/api/v1/vehicles/{vehicle_id}", status_code=204) def delete_my_vehicle(vehicle_id: uuid.UUID, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)): vehicle = db.scalar(select(Vehicle).where(Vehicle.id == vehicle_id, Vehicle.user_id == user.id)) if vehicle is None: raise HTTPException(status_code=404, detail="Vehicle not found") audit(db, "vehicle.deleted", actor_user_id=user.id, target_user_id=user.id, metadata={"vehicle_id": str(vehicle.id)}) db.delete(vehicle) db.commit() return None @app.get("/api/v1/rides") def list_my_rides(page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=100), user: User = Depends(get_current_user), db: DbSession = Depends(get_db)): query = select(Ride).where(Ride.user_id == user.id) total = db.scalar(select(func.count()).select_from(Ride).where(Ride.user_id == user.id)) or 0 rides = db.scalars(query.order_by(Ride.started_at.desc()).offset((page - 1) * page_size).limit(page_size)).all() return {"items": [ride_payload(ride) for ride in rides], "page": page, "page_size": page_size, "total": total} @app.post("/api/v1/rides", status_code=201) def create_my_ride(request: UserRideCreate, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)): if request.vehicle_id is not None: vehicle = db.scalar(select(Vehicle).where(Vehicle.id == request.vehicle_id, Vehicle.user_id == user.id)) if vehicle is None: raise HTTPException(status_code=400, detail="Vehicle not found") ride = Ride(user_id=user.id, **request.model_dump()) db.add(ride) db.flush() audit(db, "ride.created", actor_user_id=user.id, target_user_id=user.id, metadata={"ride_id": str(ride.id)}) db.commit() db.refresh(ride) return ride_payload(ride) @app.get("/api/v1/rides/{ride_id}") def get_my_ride(ride_id: uuid.UUID, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)): ride = db.scalar(select(Ride).where(Ride.id == ride_id, Ride.user_id == user.id)) if ride is None: raise HTTPException(status_code=404, detail="Ride not found") return ride_payload(ride) def telemetry_payload(point: TelemetryPoint) -> dict: return {"id": str(point.id), "ride_id": str(point.ride_id), "recorded_at": point.recorded_at, "latitude": point.latitude, "longitude": point.longitude, "speed_mph": point.speed_mph, "lean_angle_deg": point.lean_angle_deg, "g_force": point.g_force, "accel_g": point.accel_g, "brake_g": point.brake_g} def get_owned_ride(ride_id: uuid.UUID, user_id: uuid.UUID, db: DbSession) -> Ride: ride = db.scalar(select(Ride).where(Ride.id == ride_id, Ride.user_id == user_id)) if ride is None: raise HTTPException(status_code=404, detail="Ride not found") return ride @app.post("/api/v1/rides/{ride_id}/telemetry", status_code=201) def ingest_telemetry(ride_id: uuid.UUID, request: TelemetryBatch, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)): get_owned_ride(ride_id, user.id, db) points = [TelemetryPoint(ride_id=ride_id, **point.model_dump()) for point in request.points] db.add_all(points) audit(db, "telemetry.ingested", actor_user_id=user.id, target_user_id=user.id, metadata={"ride_id": str(ride_id), "point_count": len(points)}) db.commit() return {"accepted": len(points)} @app.get("/api/v1/rides/{ride_id}/telemetry") def list_my_telemetry(ride_id: uuid.UUID, limit: int = Query(default=1000, ge=1, le=5000), user: User = Depends(get_current_user), db: DbSession = Depends(get_db)): get_owned_ride(ride_id, user.id, db) points = db.scalars(select(TelemetryPoint).where(TelemetryPoint.ride_id == ride_id).order_by(TelemetryPoint.recorded_at.asc()).limit(limit)).all() return {"items": [telemetry_payload(point) for point in points], "count": len(points)} @app.get("/api/v1/admin/rides/{ride_id}/telemetry") def list_admin_telemetry(ride_id: uuid.UUID, limit: int = Query(default=1000, ge=1, le=5000), user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): if db.get(Ride, ride_id) is None: raise HTTPException(status_code=404, detail="Ride not found") points = db.scalars(select(TelemetryPoint).where(TelemetryPoint.ride_id == ride_id).order_by(TelemetryPoint.recorded_at.asc()).limit(limit)).all() return {"items": [telemetry_payload(point) for point in points], "count": len(points)} def subscription_payload(subscription: Subscription) -> dict: return {"id": str(subscription.id), "user_id": str(subscription.user_id), "plan": subscription.plan, "status": subscription.status, "current_period_start": subscription.current_period_start, "current_period_end": subscription.current_period_end, "created_at": subscription.created_at, "updated_at": subscription.updated_at} def entitlement_payload(entitlement: Entitlement) -> dict: return {"id": str(entitlement.id), "user_id": str(entitlement.user_id), "key": entitlement.key, "value": entitlement.value, "expires_at": entitlement.expires_at, "created_at": entitlement.created_at} def get_or_create_subscription(user_id: uuid.UUID, db: DbSession) -> Subscription: subscription = db.scalar(select(Subscription).where(Subscription.user_id == user_id)) if subscription is None: subscription = Subscription(user_id=user_id, plan="free", status="active") db.add(subscription) db.flush() return subscription @app.get("/api/v1/me/subscription") def get_my_subscription(user: User = Depends(get_current_user), db: DbSession = Depends(get_db)): subscription = get_or_create_subscription(user.id, db) db.commit() db.refresh(subscription) return subscription_payload(subscription) @app.get("/api/v1/me/entitlements") def list_my_entitlements(user: User = Depends(get_current_user), db: DbSession = Depends(get_db)): entitlements = db.scalars(select(Entitlement).where(Entitlement.user_id == user.id).order_by(Entitlement.created_at.desc())).all() return [entitlement_payload(item) for item in entitlements if item.expires_at is None or item.expires_at > datetime.now(timezone.utc)] @app.get("/api/v1/admin/subscriptions") def list_subscriptions(page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=100), user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): total = db.scalar(select(func.count()).select_from(Subscription)) or 0 subscriptions = db.scalars(select(Subscription).order_by(Subscription.updated_at.desc()).offset((page - 1) * page_size).limit(page_size)).all() return {"items": [subscription_payload(item) for item in subscriptions], "page": page, "page_size": page_size, "total": total} @app.patch("/api/v1/admin/users/{user_id}/subscription") def update_subscription(user_id: uuid.UUID, request: SubscriptionUpdate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): if db.get(User, user_id) is None: raise HTTPException(status_code=404, detail="User not found") subscription = get_or_create_subscription(user_id, db) subscription.plan = request.plan subscription.status = request.status subscription.current_period_start = request.current_period_start subscription.current_period_end = request.current_period_end audit(db, "admin.subscription_updated", actor_user_id=actor.id, target_user_id=user_id, metadata={"plan": request.plan, "status": request.status}) db.commit() db.refresh(subscription) return subscription_payload(subscription) @app.get("/api/v1/admin/users/{user_id}/entitlements") def list_entitlements(user_id: uuid.UUID, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): if db.get(User, user_id) is None: raise HTTPException(status_code=404, detail="User not found") return [entitlement_payload(item) for item in db.scalars(select(Entitlement).where(Entitlement.user_id == user_id).order_by(Entitlement.created_at.desc())).all()] @app.post("/api/v1/admin/users/{user_id}/entitlements", status_code=201) def create_entitlement(user_id: uuid.UUID, request: EntitlementCreate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): if db.get(User, user_id) is None: raise HTTPException(status_code=404, detail="User not found") entitlement = Entitlement(user_id=user_id, **request.model_dump()) db.add(entitlement) db.flush() audit(db, "admin.entitlement_created", actor_user_id=actor.id, target_user_id=user_id, metadata={"key": entitlement.key}) db.commit() db.refresh(entitlement) return entitlement_payload(entitlement)