Urban match python task

database.py

@@ -1,11 +1,8 @@
 from sqlalchemy import create_engine
-from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import sessionmaker, declarative_base  # Updated import
 
-// Free to use remote db or create a local database. Modify the URl appropriately
 SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
 
-engine = create_engine(SQLALCHEMY_DATABASE_URL)
+engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
-Base = declarative_base()
-
+Base = declarative_base()  

main.py

@@ -1,3 +1,4 @@
+import json
 from fastapi import FastAPI, HTTPException, Depends
 from sqlalchemy.orm import Session
 from database import SessionLocal, engine, Base
@@ -16,21 +17,86 @@ def get_db():
 
 @app.post("/users/", response_model=schemas.User)
 def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
-    db_user = models.User(**user.dict())
+    existing_user = db.query(models.User).filter(models.User.email == user.email).first()
+    if existing_user:
+        raise HTTPException(status_code=400, detail="Email already registered")
+
+    db_user = models.User(
+        **user.model_dump(exclude={"interests"}),  
+        interests=json.dumps(user.interests)  
+    )
     db.add(db_user)
     db.commit()
     db.refresh(db_user)
+    db_user.interests = json.loads(db_user.interests)  
     return db_user
 
 @app.get("/users/", response_model=list[schemas.User])
 def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
-    users = db.query(models.User).offset(skip).limit(limit).all()
+    users = db.query(models.User).filter(models.User.is_deleted == False).offset(skip).limit(limit).all()
+
+    for user in users:
+        user.interests = json.loads(user.interests) if user.interests else []
+
     return users
 
 @app.get("/users/{user_id}", response_model=schemas.User)
 def read_user(user_id: int, db: Session = Depends(get_db)):
-    user = db.query(models.User).filter(models.User.id == user_id).first()
+    user = db.query(models.User).filter(models.User.id == user_id, models.User.is_deleted == False).first()
     if user is None:
         raise HTTPException(status_code=404, detail="User not found")
+
+    user.interests = json.loads(user.interests) if user.interests else []
+
     return user
 
+@app.get("/users/{user_id}/matches", response_model=list[schemas.User])
+def find_matches(user_id: int, db: Session = Depends(get_db)):
+    db_user = db.query(models.User).filter(models.User.id == user_id, models.User.is_deleted == False).first()
+    if db_user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    user_interests = json.loads(db_user.interests) if db_user.interests else []
+    all_users = db.query(models.User).filter(models.User.is_deleted == False, models.User.id != user_id).all()
+
+    matches = [
+        user for user in all_users
+        if set(json.loads(user.interests) if user.interests else []).intersection(user_interests)
+    ]
+
+    for match in matches:
+        match.interests = json.loads(match.interests) if match.interests else []
+
+    return matches
+
+@app.put("/users/{user_id}", response_model=schemas.User)
+def update_user(user_id: int, user_update: schemas.UserUpdate, db: Session = Depends(get_db)):
+    db_user = db.query(models.User).filter(models.User.id == user_id, models.User.is_deleted == False).first()
+    if db_user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    update_data = user_update.model_dump(exclude_unset=True)  
+
+    if "email" in update_data and not schemas.validate_email(update_data["email"]):
+        raise HTTPException(status_code=400, detail="Invalid email format")
+
+    if "interests" in update_data:
+        update_data["interests"] = json.dumps(update_data["interests"])  
+
+    for key, value in update_data.items():
+        setattr(db_user, key, value)
+
+    db.commit()
+    db.refresh(db_user)
+    db_user.interests = json.loads(db_user.interests) if db_user.interests else []
+    return db_user
+
+@app.delete("/users/{user_id}")
+def delete_user(user_id: int, db: Session = Depends(get_db)):
+    db_user = db.query(models.User).filter(models.User.id == user_id).first()
+    if db_user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    db_user.is_deleted = True  
+    db.commit()
+    return {"message": "User marked as deleted (soft delete)."}

models.py

@@ -1,5 +1,6 @@
-from sqlalchemy import Column, Integer, String, ARRAY
+from sqlalchemy import Column, Integer, String, Boolean, Text
 from database import Base
+import json
 
 class User(Base):
     __tablename__ = "users"
@@ -10,5 +11,11 @@ class User(Base):
     gender = Column(String)
     email = Column(String, unique=True, index=True)
     city = Column(String, index=True)
-    interests = Column(ARRAY(String))
+    interests = Column(Text, nullable=True) 
+    is_deleted = Column(Boolean, default=False)
 
+    def set_interests(self, interests_list):
+        self.interests = json.dumps(interests_list)
+
+    def get_interests(self):
+        return json.loads(self.interests) if self.interests else []

schemas.py

@@ -1,20 +1,35 @@
-from pydantic import BaseModel
-from typing import List
+from pydantic import BaseModel, EmailStr, ValidationError
+from typing import List, Optional
 
 class UserBase(BaseModel):
+    name: Optional[str] = None
+    age: Optional[int] = None
+    gender: Optional[str] = None
+    email: Optional[EmailStr] = None
+    city: Optional[str] = None
+    interests: Optional[List[str]] = None 
+
+class UserCreate(UserBase):
     name: str
     age: int
     gender: str
-    email: str
+    email: EmailStr
     city: str
-    interests: List[str]
+    interests: List[str]  
 
-class UserCreate(UserBase):
+class UserUpdate(UserBase): 
     pass
 
 class User(UserBase):
     id: int
 
     class Config:
-        orm_mode = True
+        from_attributes = True  
 
+def validate_email(email: str) -> bool:
+    """Validate email format manually."""
+    try:
+        EmailStr.validate(email)
+        return True
+    except ValidationError:
+        return False