Building a Freelancing Platform with FastAPI and MongoDB

Mosta9el is a freelancing platform I built from scratch using FastAPI for the backend, MongoDB for the database, and Flutter for the mobile client. Here's the full technical breakdown.

Why FastAPI + MongoDB?

The requirements were clear: fast API responses for a mobile-first experience, flexible data model for job listings (which vary wildly in structure), and quick iteration speed. FastAPI gave us automatic OpenAPI docs, async support out of the box, and Pydantic for validation. MongoDB's document model handled the variability in job/proposal structures without painful schema migrations.

Project Structure

mosta9el-api/
├── app/
│   ├── core/
│   │   ├── config.py      # Settings via pydantic-settings
│   │   ├── database.py    # MongoDB connection
│   │   └── security.py    # JWT utilities
│   ├── models/            # Pydantic models (request/response)
│   ├── repositories/      # DB access layer
│   ├── routers/           # Route handlers
│   └── services/          # Business logic
├── tests/
└── main.py

Database Layer

I use Motor (async MongoDB driver) with a simple repository pattern:

# app/core/database.py
from motor.motor_asyncio import AsyncIOMotorClient
from beanie import init_beanie
from app.models.user import User
from app.models.job import Job
 
client: AsyncIOMotorClient = None
 
async def connect_db():
    global client
    client = AsyncIOMotorClient(settings.MONGODB_URL)
    await init_beanie(
        database=client[settings.DB_NAME],
        document_models=[User, Job, Proposal, Review],
    )
 
async def close_db():
    client.close()

I use Beanie as an ODM — it sits on top of Motor and gives you a clean model-based API similar to Mongoose but Pythonic.

# app/models/job.py
from beanie import Document
from pydantic import Field
from typing import Optional, List
from datetime import datetime
from enum import Enum
 
class JobStatus(str, Enum):
    OPEN = "open"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    CANCELLED = "cancelled"
 
class Job(Document):
    title: str
    description: str
    budget_min: float
    budget_max: float
    skills_required: List[str] = []
    status: JobStatus = JobStatus.OPEN
    client_id: str
    
    # MongoDB will store these as nested documents
    category: str
    location: Optional[str] = None
    deadline: Optional[datetime] = None
    
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)
 
    class Settings:
        name = "jobs"
        indexes = [
            "status",
            "client_id",
            [("skills_required", 1), ("status", 1)],
            [("created_at", -1)],
        ]

Authentication with JWT

# app/core/security.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
 
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")
 
# Dependency for protected routes
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    credentials_exception = HTTPException(
        status_code=401,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
        user_id: str = payload.get("sub")
        if not user_id:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    user = await User.get(user_id)
    if not user:
        raise credentials_exception
    return user

Job Search with Text + Filters

MongoDB's compound index and aggregation pipeline make filtering efficient:

@router.get("/jobs")
async def list_jobs(
    q: Optional[str] = None,
    skills: Optional[List[str]] = Query(None),
    min_budget: Optional[float] = None,
    max_budget: Optional[float] = None,
    page: int = 1,
    limit: int = 20,
):
    query = {"status": JobStatus.OPEN}
    
    if q:
        query["$text"] = {"$search": q}
    if skills:
        query["skills_required"] = {"$all": skills}
    if min_budget is not None:
        query["budget_max"] = {"$gte": min_budget}
    if max_budget is not None:
        query["budget_min"] = {"$lte": max_budget}
    
    total = await Job.find(query).count()
    jobs = await Job.find(query)\
        .sort(-Job.created_at)\
        .skip((page - 1) * limit)\
        .limit(limit)\
        .to_list()
    
    return {
        "jobs": jobs,
        "total": total,
        "page": page,
        "pages": (total + limit - 1) // limit,
    }

Flutter API Client

On the Flutter side, I used Dio with an interceptor for auth:

class ApiClient {
  late final Dio _dio;
 
  ApiClient() {
    _dio = Dio(BaseOptions(baseUrl: Env.apiUrl));
    _dio.interceptors.add(AuthInterceptor());
  }
}
 
class AuthInterceptor extends Interceptor {
  @override
  void onRequest(RequestOptions options, RequestInterceptorHandler handler) {
    final token = SecureStorage.read('access_token');
    if (token != null) {
      options.headers['Authorization'] = 'Bearer $token';
    }
    handler.next(options);
  }
 
  @override
  void onError(DioException err, ErrorInterceptorHandler handler) async {
    if (err.response?.statusCode == 401) {
      // Refresh token and retry
      final refreshed = await _refreshToken();
      if (refreshed) {
        handler.resolve(await _retry(err.requestOptions));
        return;
      }
    }
    handler.next(err);
  }
}

Performance Optimizations

Projection — never fetch the full document when you only need a subset:

jobs = await Job.find(query, projection_model=JobSummary).to_list()

Indexes — the compound index on (skills_required, status) reduced job search query time from ~200ms to ~8ms.
Connection pooling — Motor handles this, but set maxPoolSize appropriately for your server's memory.
Response caching — job category pages are cached in Redis for 60 seconds.

FastAPI + MongoDB is a genuinely good combination for mobile-first platforms where flexibility and speed matter more than strict schema enforcement.