Add routes and Test

Author: AliiAhmadi

.env.example

@@ -2,3 +2,4 @@ DB_NAME=fastapi
 DB_HOST=localhost
 DB_USER=root
 DB_PASSWORD=
+DB_PORT=5432

app/crud.py

@@ -1,31 +1,17 @@
 from sqlalchemy import create_engine
-from dotenv import load_dotenv
 from sqlalchemy.orm import sessionmaker
-from sqlalchemy.ext.declarative import declarative_base
-import os
 
-load_dotenv()
 
-
-DB_NAME = os.getenv("DB_NAME")
-DB_USER = os.getenv("DB_USER")
-DB_PASSWORD = os.getenv("DB_PASSWORD")
-DB_HOST = os.getenv("DB_HOST")
-
-DATABASE_URL = f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}/{DB_NAME}"
+DATABASE_URL = "sqlite:///../users.db"
 
 engine = create_engine(DATABASE_URL)
 
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
 
-Base = declarative_base()
-
-
 def get_db():
     db = SessionLocal()
     try:
         yield db
     finally:
-        db.close()
-    
+        db.close()

app/database.py

@@ -1,19 +1,49 @@
-from fastapi import FastAPI
-from .routers.user import router
-from .database import engine
+from fastapi import FastAPI, HTTPException, Depends
+from sqlalchemy.orm import Session
+from database import get_db
+from models import User
+from schema import UserCreate, UserUpdate
+
 
 
 app = FastAPI()
 
-app.include_router(router)
 
-#===========================#
+@app.get("/users/")
+def get_all_users(db: Session = Depends(get_db)):
+    return db.query(User).all()
+
+@app.get("/users/{user_id}")
+def get_user_by_email(user_id: int, db: Session = Depends(get_db)):
+    user = db.query(User).filter(User.id == user_id).first()
+    if user:
+        return user
+    raise HTTPException(status_code=404, detail="User not found")
+
+
+@app.post("/users/")
+def create_user(user: UserCreate, db: Session = Depends(get_db)):
+    db_user = User(name=user.name, email=user.email, password=user.password)
+    db.add(db_user)
+    db.commit()
+    db.refresh(db_user)
+    return db_user
 
-@app.on_event("startup")
-async def startup():
-    await engine.connect()
-    
+@app.put("/users/{user_id}")
+def update_user_by_email(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):
+    db_user = db.query(User).filter(User.id == user_id).first()
+    if not db_user:
+        raise HTTPException(status_code=404, detail="User not found")
+    db_user.name = user.name
+    db_user.email = user.email
+    db.commit()
+    return {"message": "User updated successfully"}
 
-@app.on_event("shutdown")
-async def shutdown():
-    await engine.disconnect()
+@app.delete("/users/{user_id}")
+def delete_user_by_email(user_id: int, db: Session = Depends(get_db)):
+    db_user = db.query(User).filter(User.id == user_id).first()
+    if not db_user:
+        raise HTTPException(status_code=404, detail="User not found")
+    db.delete(db_user)
+    db.commit()
+    return {"message": "User deleted successfully"}

app/main.py

@@ -1,23 +1,11 @@
-from pydantic import BaseModel
-from .database import Base
-from sqlalchemy import Column, String
+from sqlalchemy import Column, Integer, String
+from sqlalchemy.ext.declarative import declarative_base
 
-class UserCreate(BaseModel):
-    name: str
-    email: str
-    password: str
-    
-
-class UserUpdate(BaseModel):
-    name: str
-    email: str
-    password: str
-    
 
+Base = declarative_base()
 class User(Base):
     __tablename__ = "users"
-    
-    id = Column(primary_key=True)
-    name = Column(String, nullable=False)
-    email = Column(String, nullable=False)
-    password = Column(String, nullable=False)
+    id: int = Column(Integer, primary_key=True, index=True)
+    name: str = Column(String(50), index=True)
+    email: str = Column(String(50), unique=True, index=True)
+    password: str = Column(String(50))

app/models.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel
+
+
+class UserCreate(BaseModel):
+    name: str
+    email: str
+    password: str
+
+class UserUpdate(BaseModel):
+    name: str
+    email: str