Python Task Submission: Akash Chattopadhyay

.gitignore

@@ -0,0 +1 @@
+**/__pycache__/

db/config.py

@@ -0,0 +1,19 @@
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+# Free to use remote db or create a local database. Modify the URl appropriately
+# The use of Arrays in data only works in Postgres.
+SQLALCHEMY_DATABASE_URL = "postgresql://user:password@host/dbName?sslmode=require"
+
+engine = create_engine(SQLALCHEMY_DATABASE_URL)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+Base = declarative_base()
+Base.metadata.create_all(bind=engine)
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

main.py

@@ -1,36 +1,100 @@
+# Suggested code may be subject to a license. Learn more: ~LicenseLog:140021096.
 from fastapi import FastAPI, HTTPException, Depends
+from sqlalchemy import text
 from sqlalchemy.orm import Session
-from database import SessionLocal, engine, Base
-import models, schemas
+from db.config import get_db
+import models
+import schemas.UserSchema as UserSchemas
+from utils.validation import email_validation
 
 app = FastAPI()
 
-Base.metadata.create_all(bind=engine)
-
-def get_db():
-    db = SessionLocal()
-    try:
-        yield db
-    finally:
-        db.close()
-
-@app.post("/users/", response_model=schemas.User)
-def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
+@app.post("/users/", response_model=UserSchemas.User)
+def create_user(user: UserSchemas.UserCreate, db: Session = Depends(get_db)):
+    if not email_validation(user.email):
+        raise HTTPException(status_code=400, detail="Invalid email address")
+    db_user = db.query(models.User).filter(models.User.email == user.email).first()
+    if db_user:
+        raise HTTPException(status_code=400, detail="Email already registered")
     db_user = models.User(**user.dict())
     db.add(db_user)
     db.commit()
     db.refresh(db_user)
     return db_user
 
-@app.get("/users/", response_model=list[schemas.User])
+@app.get("/users/", response_model=list[UserSchemas.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()
     return users
 
-@app.get("/users/{user_id}", response_model=schemas.User)
+@app.get("/users/{user_id}", response_model=UserSchemas.User)
 def read_user(user_id: int, db: Session = Depends(get_db)):
     user = db.query(models.User).filter(models.User.id == user_id).first()
     if user is None:
         raise HTTPException(status_code=404, detail="User not found")
     return user
 
+@app.patch("/users/{user_id}", response_model=UserSchemas.User)
+def update_user(user_id: int, user: UserSchemas.UserUpdate, 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")
+    update_data = user.dict(exclude_unset=True)
+    for key, value in update_data.items():
+        setattr(db_user, key, value)
+    db.commit()
+    db.refresh(db_user)
+    return db_user
+
+@app.delete("/users/{user_id}", response_model=UserSchemas.User)
+def delete_user(user_id: int, db: Session = Depends(get_db)):
+    user = db.query(models.User).filter(models.User.id == user_id).first()
+    if user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+    db.delete(user)
+    db.commit()
+    return user
+
+@app.get("/recommendation/{user_id}", response_model=list[UserSchemas.User])
+def recommendation(user_id: int, db: Session = Depends(get_db)):
+    user = db.query(models.User).filter(models.User.id == user_id).first()
+    if user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    query = text("""
+        SELECT 
+            id, name, age, gender, email, city, interests,
+            array_length(ARRAY(
+                SELECT unnest(interests) 
+                INTERSECT 
+                SELECT unnest(ARRAY[:user_interests]::VARCHAR[])
+            ), 1) AS intersection_size
+        FROM users
+        WHERE 
+            id != :user_id AND
+            city = :city AND
+            age BETWEEN :min_age AND :max_age AND
+            (
+                (age >= 18 AND :user_age >= 18) OR 
+                (age < 18 AND :user_age < 18)
+            ) AND
+            interests && ARRAY[:user_interests]::VARCHAR[]
+        ORDER BY intersection_size DESC
+    """)
+
+    # Replace these with your actual variables
+    params = {
+        "user_id": user_id,
+        "city": user.city,
+        "min_age": user.age - 2,
+        "max_age": user.age + 2,
+        "user_age": user.age,
+        "user_interests": user.interests,
+    }
+
+    recommendations = db.execute(query, params).fetchall()
+
+    return recommendations
+
+if __name__ == "__main__":
+    app.run()

models.py → models/UserModel.py

@@ -1,5 +1,5 @@
 from sqlalchemy import Column, Integer, String, ARRAY
-from database import Base
+from db.config import Base
 
 class User(Base):
     __tablename__ = "users"

models/__init__.py

@@ -0,0 +1 @@
+from .UserModel import User

requirements.txt

@@ -0,0 +1,15 @@
+annotated-types==0.7.0
+anyio==4.8.0
+click==8.1.8
+fastapi==0.115.6
+greenlet==3.1.1
+h11==0.14.0
+idna==3.10
+psycopg2-binary==2.9.10
+pydantic==2.10.4
+pydantic_core==2.27.2
+sniffio==1.3.1
+SQLAlchemy==2.0.36
+starlette==0.41.3
+typing_extensions==4.12.2
+uvicorn==0.34.0

schemas.py → schemas/UserSchema.py

@@ -12,9 +12,11 @@ class UserBase(BaseModel):
 class UserCreate(UserBase):
     pass
 
+class UserUpdate(UserBase):
+    pass
+
 class User(UserBase):
     id: int
 
     class Config:
-        orm_mode = True
-
+        from_attributes = True

test.sh

@@ -0,0 +1,67 @@
+BaseURL="http://127.0.0.1:8000/"
+BaseUserURL="${BaseURL}users/"
+BaseRecommendationURL="${BaseURL}recommendation/"
+
+all_users=`curl ${BaseUserURL} -s`
+user_count_1=$(echo "${all_users}" | jq '. | length')
+
+email=$(echo "akash$(shuf -i 1-1000000 -n 1)-$(date +%s)"@gmail.com)
+user='{"name": "Akash Chattopadhyay", "age": 24, "gender": "male", "email": "'"${email}"'", "city": "Gurgaon", "interests": ["sports"]}'
+new_user=`curl -s -H 'Content-Type: application/json' -d "${user}" -X POST ${BaseUserURL}`
+all_users=`curl ${BaseUserURL} -s`
+user_count_2=$(echo "${all_users}" | jq '. | length')
+
+if [[ $user_count_1 != $((user_count_2 - 1)) ]]
+then
+    echo "Things went wrong in testing positive for creation"
+    exit 0
+fi
+
+new_user_1=`curl -s -H 'Content-Type: application/json' -d "${user}" -X POST ${BaseUserURL}`
+all_users=`curl ${BaseUserURL} -s`
+user_count_2=$(echo "${all_users}" | jq '. | length')
+if [[ $user_count_1 == $((user_count_2 - 2)) ]]
+then
+    echo "Things went wrong in testing negative for creation"
+    exit 0
+fi
+
+id=$(echo "${new_user}" | jq '.id')
+final_user_url="${BaseUserURL}${id}"
+user_test=`curl -s -H 'Content-Type: application/json' -X GET ${final_user_url}`
+
+if [[ "${user_test}" != "${new_user}" ]]
+then
+    echo "Things went wrong in testing positive for reading one"
+    exit 0
+fi
+
+user_test=`curl -s -H 'Content-Type: application/json' -X GET ${BaseUserURL}/-1`
+if [[ "${user_test}" == "${new_user}" ]]
+then
+    echo "Things went wrong in testing negative for reading one"
+    exit 0
+fi
+
+user='{"name": "John Doe", "age": 24, "gender": "male", "email": "'"${email}"'", "city": "Gurgaon", "interests": ["reading"]}'
+user_test_2=`curl -s -H 'Content-Type: application/json' -d "${user}" -X PATCH ${final_user_url}`
+user_test=`curl -s -H 'Content-Type: application/json' -X GET ${final_user_url}`
+
+if [[ "$(echo "${user_test}" | jq '.name')" != "$(echo "${user}" | jq '.name')" ]]
+then
+    echo "Things went wrong in testing positive for updating"
+    exit 0
+fi
+
+id=$(echo "${user_test}" | jq '.id')
+recommendations=`curl -s -H 'Content-Type: application/json' -X GET ${BaseRecommendationURL}${id}`
+user_count_1=$(echo "${recommendations}" | jq '. | length')
+echo $user_count_1
+
+user_test=`curl -s -H 'Content-Type: application/json' -X DELETE ${final_user_url}`
+user_test_2=`curl -s -H 'Content-Type: application/json' -X GET ${final_user_url}`
+if [[ "${user_test_2}" == "${user_test}" ]]
+then
+    echo "Things went wrong in testing positive for deleting"
+    exit 0
+fi

utils/validation.py

@@ -0,0 +1,8 @@
+import re
+
+def email_validation(email: str) -> bool:
+    regex = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b'
+    if(re.fullmatch(regex, email)):
+        return True
+    else:
+        return False