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
137 lines
4.3 KiB
Python
137 lines
4.3 KiB
Python
from typing import List, Optional
|
|
from fastapi import APIRouter, Depends, HTTPException, status, Query
|
|
from sqlalchemy.orm import Session
|
|
from app.auth.dependencies import get_db, get_current_active_user
|
|
from app.models.user import User
|
|
from app.models.property import PropertyListing, PropertyType, PropertyStatus
|
|
from app.schemas.property import PropertyCreate, PropertyUpdate, PropertyResponse
|
|
|
|
router = APIRouter(prefix="/api/properties", tags=["Properties"])
|
|
|
|
|
|
@router.post("/", response_model=PropertyResponse)
|
|
def create_property(
|
|
property_data: PropertyCreate,
|
|
current_user: User = Depends(get_current_active_user),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
db_property = PropertyListing(
|
|
owner_id=current_user.id,
|
|
**property_data.dict()
|
|
)
|
|
db.add(db_property)
|
|
db.commit()
|
|
db.refresh(db_property)
|
|
return db_property
|
|
|
|
|
|
@router.get("/", response_model=List[PropertyResponse])
|
|
def get_properties(
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(100, ge=1, le=100),
|
|
location: Optional[str] = Query(None),
|
|
state: Optional[str] = Query(None),
|
|
property_type: Optional[PropertyType] = Query(None),
|
|
status: Optional[PropertyStatus] = Query(None),
|
|
min_price: Optional[float] = Query(None, ge=0),
|
|
max_price: Optional[float] = Query(None, ge=0),
|
|
bedrooms: Optional[int] = Query(None, ge=0),
|
|
is_affordable: Optional[bool] = Query(None),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
query = db.query(PropertyListing).filter(
|
|
PropertyListing.is_active,
|
|
PropertyListing.is_approved
|
|
)
|
|
|
|
if location:
|
|
query = query.filter(PropertyListing.location.ilike(f"%{location}%"))
|
|
|
|
if state:
|
|
query = query.filter(PropertyListing.state.ilike(f"%{state}%"))
|
|
|
|
if property_type:
|
|
query = query.filter(PropertyListing.property_type == property_type)
|
|
|
|
if status:
|
|
query = query.filter(PropertyListing.status == status)
|
|
|
|
if min_price is not None:
|
|
query = query.filter(PropertyListing.price >= min_price)
|
|
|
|
if max_price is not None:
|
|
query = query.filter(PropertyListing.price <= max_price)
|
|
|
|
if bedrooms is not None:
|
|
query = query.filter(PropertyListing.bedrooms >= bedrooms)
|
|
|
|
if is_affordable is not None:
|
|
query = query.filter(PropertyListing.is_affordable == is_affordable)
|
|
|
|
properties = query.offset(skip).limit(limit).all()
|
|
return properties
|
|
|
|
|
|
@router.get("/{property_id}", response_model=PropertyResponse)
|
|
def get_property(property_id: int, db: Session = Depends(get_db)):
|
|
property_listing = db.query(PropertyListing).filter(
|
|
PropertyListing.id == property_id,
|
|
PropertyListing.is_active
|
|
).first()
|
|
|
|
if not property_listing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Property not found"
|
|
)
|
|
|
|
return property_listing
|
|
|
|
|
|
@router.put("/{property_id}", response_model=PropertyResponse)
|
|
def update_property(
|
|
property_id: int,
|
|
property_data: PropertyUpdate,
|
|
current_user: User = Depends(get_current_active_user),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
property_listing = db.query(PropertyListing).filter(
|
|
PropertyListing.id == property_id,
|
|
PropertyListing.owner_id == current_user.id
|
|
).first()
|
|
|
|
if not property_listing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Property not found or not owned by you"
|
|
)
|
|
|
|
update_data = property_data.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(property_listing, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(property_listing)
|
|
return property_listing
|
|
|
|
|
|
@router.delete("/{property_id}")
|
|
def delete_property(
|
|
property_id: int,
|
|
current_user: User = Depends(get_current_active_user),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
property_listing = db.query(PropertyListing).filter(
|
|
PropertyListing.id == property_id,
|
|
PropertyListing.owner_id == current_user.id
|
|
).first()
|
|
|
|
if not property_listing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Property not found or not owned by you"
|
|
)
|
|
|
|
property_listing.is_active = False
|
|
db.commit()
|
|
return {"message": "Property deleted successfully"} |