9ca9d35a1a
- Set up FastAPI project structure with modular architecture - Create comprehensive database models for users, properties, messages, notifications, and payments - Implement JWT-based authentication with role-based access control (Seeker, Agent, Landlord, Admin) - Build property listings CRUD with advanced search and filtering capabilities - Add dedicated affordable housing endpoints for Nigerian market focus - Create real-time messaging system between users - Implement admin dashboard with property approval workflow and analytics - Add notification system for user alerts - Integrate Paystack payment gateway for transactions - Set up SQLite database with Alembic migrations - Include comprehensive health check and API documentation - Add proper error handling and validation throughout - Follow FastAPI best practices with Pydantic schemas and dependency injection
144 lines
4.1 KiB
Python
144 lines
4.1 KiB
Python
from typing import List
|
|
from fastapi import APIRouter, Depends, HTTPException, status, Query
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import func
|
|
from app.auth.dependencies import get_db, get_current_active_user
|
|
from app.models.user import User, UserRole
|
|
from app.models.property import PropertyListing
|
|
from app.models.message import Message
|
|
from app.schemas.property import PropertyResponse
|
|
from app.schemas.user import UserResponse
|
|
|
|
router = APIRouter(prefix="/api/admin", tags=["Admin"])
|
|
|
|
|
|
def get_admin_user(current_user: User = Depends(get_current_active_user)) -> User:
|
|
if current_user.role != UserRole.ADMIN:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Admin access required"
|
|
)
|
|
return current_user
|
|
|
|
|
|
@router.get("/properties/pending", response_model=List[PropertyResponse])
|
|
def get_pending_properties(
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(50, ge=1, le=100),
|
|
admin_user: User = Depends(get_admin_user),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
properties = db.query(PropertyListing).filter(
|
|
~PropertyListing.is_approved,
|
|
PropertyListing.is_active
|
|
).offset(skip).limit(limit).all()
|
|
|
|
return properties
|
|
|
|
|
|
@router.put("/properties/{property_id}/approve")
|
|
def approve_property(
|
|
property_id: int,
|
|
admin_user: User = Depends(get_admin_user),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
property_listing = db.query(PropertyListing).filter(
|
|
PropertyListing.id == property_id
|
|
).first()
|
|
|
|
if not property_listing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Property not found"
|
|
)
|
|
|
|
property_listing.is_approved = True
|
|
db.commit()
|
|
|
|
return {"message": "Property approved successfully"}
|
|
|
|
|
|
@router.put("/properties/{property_id}/reject")
|
|
def reject_property(
|
|
property_id: int,
|
|
admin_user: User = Depends(get_admin_user),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
property_listing = db.query(PropertyListing).filter(
|
|
PropertyListing.id == property_id
|
|
).first()
|
|
|
|
if not property_listing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Property not found"
|
|
)
|
|
|
|
property_listing.is_active = False
|
|
db.commit()
|
|
|
|
return {"message": "Property rejected successfully"}
|
|
|
|
|
|
@router.get("/users", response_model=List[UserResponse])
|
|
def get_users(
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(50, ge=1, le=100),
|
|
role: UserRole = Query(None),
|
|
admin_user: User = Depends(get_admin_user),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
query = db.query(User)
|
|
|
|
if role:
|
|
query = query.filter(User.role == role)
|
|
|
|
users = query.offset(skip).limit(limit).all()
|
|
return users
|
|
|
|
|
|
@router.put("/users/{user_id}/deactivate")
|
|
def deactivate_user(
|
|
user_id: int,
|
|
admin_user: User = Depends(get_admin_user),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="User not found"
|
|
)
|
|
|
|
if user.role == UserRole.ADMIN:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Cannot deactivate admin user"
|
|
)
|
|
|
|
user.is_active = False
|
|
db.commit()
|
|
|
|
return {"message": "User deactivated successfully"}
|
|
|
|
|
|
@router.get("/analytics")
|
|
def get_analytics(
|
|
admin_user: User = Depends(get_admin_user),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
total_users = db.query(func.count(User.id)).scalar()
|
|
total_properties = db.query(func.count(PropertyListing.id)).scalar()
|
|
pending_properties = db.query(func.count(PropertyListing.id)).filter(
|
|
~PropertyListing.is_approved,
|
|
PropertyListing.is_active
|
|
).scalar()
|
|
total_messages = db.query(func.count(Message.id)).scalar()
|
|
|
|
return {
|
|
"total_users": total_users,
|
|
"total_properties": total_properties,
|
|
"pending_properties": pending_properties,
|
|
"total_messages": total_messages
|
|
} |