43235eb604
- Setup project structure and basic FastAPI application - Define database models for users, profiles, matches, and messages - Set up database connection and create Alembic migrations - Implement user authentication and registration endpoints - Create API endpoints for profile management, matches, and messaging - Add filtering and search functionality for tech profiles - Setup environment variable configuration - Create README with project information and setup instructions
135 lines
4.0 KiB
Python
135 lines
4.0 KiB
Python
from typing import Any, List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import get_current_active_user
|
|
from app.db.session import get_db
|
|
from app.models.user import User
|
|
from app.schemas.profile import Profile, ProfileCreate, ProfileUpdate
|
|
from app.crud import profile
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/me", response_model=Profile)
|
|
def read_own_profile(
|
|
current_user: User = Depends(get_current_active_user),
|
|
db: Session = Depends(get_db),
|
|
) -> Any:
|
|
"""
|
|
Get current user's profile
|
|
"""
|
|
user_profile = profile.get_by_user_id(db, user_id=current_user.id)
|
|
if not user_profile:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Profile not found",
|
|
)
|
|
return user_profile
|
|
|
|
|
|
@router.post("/", response_model=Profile)
|
|
def create_profile(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
profile_in: ProfileCreate,
|
|
current_user: User = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Create new profile for current user
|
|
"""
|
|
# Check if user already has a profile
|
|
user_profile = profile.get_by_user_id(db, user_id=current_user.id)
|
|
if user_profile:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="User already has a profile",
|
|
)
|
|
|
|
# Create profile with current user id
|
|
profile_data = profile_in.dict()
|
|
profile_data["user_id"] = current_user.id
|
|
|
|
return profile.create(db, obj_in=ProfileCreate(**profile_data))
|
|
|
|
|
|
@router.put("/me", response_model=Profile)
|
|
def update_own_profile(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
profile_in: ProfileUpdate,
|
|
current_user: User = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Update current user's profile
|
|
"""
|
|
user_profile = profile.get_by_user_id(db, user_id=current_user.id)
|
|
if not user_profile:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Profile not found",
|
|
)
|
|
|
|
updated_profile = profile.update(db, db_obj=user_profile, obj_in=profile_in)
|
|
return updated_profile
|
|
|
|
|
|
@router.get("/{profile_id}", response_model=Profile)
|
|
def read_profile(
|
|
*,
|
|
profile_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Get profile by ID
|
|
"""
|
|
user_profile = profile.get(db, id=profile_id)
|
|
if not user_profile:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Profile not found",
|
|
)
|
|
|
|
# Only show visible profiles unless it's the user's own profile
|
|
if not user_profile.is_visible and user_profile.user_id != current_user.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Profile not found",
|
|
)
|
|
|
|
return user_profile
|
|
|
|
|
|
@router.get("/", response_model=List[Profile])
|
|
def list_profiles(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_active_user),
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
tech_stack: Optional[str] = Query(None, description="Filter by tech stack (comma separated)"),
|
|
) -> Any:
|
|
"""
|
|
List all visible profiles with optional tech stack filtering
|
|
"""
|
|
# Basic query for visible profiles
|
|
if tech_stack:
|
|
# Simple filtering based on tech stack
|
|
# In a real app, you'd implement more sophisticated filtering
|
|
tech_list = [tech.strip().lower() for tech in tech_stack.split(",")]
|
|
profiles = []
|
|
all_profiles = profile.get_visible_profiles(db, skip=skip, limit=limit)
|
|
|
|
for p in all_profiles:
|
|
if not p.tech_stack:
|
|
continue
|
|
|
|
profile_tech = [tech.strip().lower() for tech in p.tech_stack.split(",")]
|
|
if any(tech in profile_tech for tech in tech_list):
|
|
profiles.append(p)
|
|
else:
|
|
profiles = profile.get_visible_profiles(db, skip=skip, limit=limit)
|
|
|
|
return profiles |