Build a Job Aggregator API with FastAPI
Build a REST API that fetches, stores, and filters real job listings.
Introduction
30 Second Summary
Searching for jobs means checking dozens of platforms, each with different filters, layouts, and data formats. What if you could pull listings from a major aggregator into your own database and query them exactly how you want?
In this project, you will build a Job Aggregator API that fetches real job listings from Adzuna, stores them in PostgreSQL, and serves them through clean, filterable endpoints designed for a mobile app to consume.
What You'll Build
A REST API built with FastAPI that integrates the Adzuna job listings API, normalizes data into PostgreSQL, and exposes paginated, filterable search endpoints.
By the end of this project, you'll have:
- A running FastAPI backend that pulls job listings from the Adzuna API and stores them in a local PostgreSQL database.
- A comprehensive filtering system with endpoints for searching jobs by keyword, category, location, salary range, and contract type.
- Auto-generated interactive API documentation via Swagger UI, ready for a Flutter mobile app to consume.
- Secret Mission: Add a job categories endpoint and a saved-jobs bookmarking system.
Are there any prerequisites?
You should have basic Python knowledge and be comfortable running commands in a terminal. You also need PostgreSQL and uv installed locally before starting.
Set Up the Project
Before you can build an API, you need a solid foundation. That means a project directory, the right packages installed, and credentials for the external data source your API will pull from.
In this step, you will initialize a Python project with uv, install your dependencies, register for a free Adzuna API key, and scaffold the file structure that FastAPI expects.
In this step, get ready to:
- Initialize the project with uv and install dependencies.
- Register for a free Adzuna API key and configure environment variables.
- Create the project file structure and verify the server runs.
Initialize the project with uv and install dependencies
uv is a fast Python package manager that replaces pip and venv with a single tool. It handles project creation, dependency resolution, and virtual environment management all in one command.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal and press Enter to open it.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project by running this command:
uv init job-aggregator-apiWhat does this command do?
uv init job-aggregator-api creates a new directory called job-aggregator-api with a pyproject.toml, a sample main.py, a README.md, and a .python-version file. It also initializes a Git repository.
- Move into your new project folder by running:
cd job-aggregator-api- Delete the default main.py file that uv created. You will create your own project structure shortly:
rm main.py- Install all the project dependencies by running this command:
uv add "fastapi[standard]>=0.136.0,<0.137.0" "sqlalchemy>=2.0.50,<2.1.0" "psycopg2-binary>=2.9.12,<3.0.0" "httpx>=0.28.1,<0.29.0" "pydantic-settings>=2.0.0,<3.0.0"What does this command do?
uv add installs packages and records them in your pyproject.toml. It also creates a virtual environment (.venv) and a lockfile (uv.lock) automatically.
- fastapi[standard] is the web framework with its dev server included.
- sqlalchemy is the ORM for interacting with your PostgreSQL database.
- psycopg2-binary is the PostgreSQL database adapter.
- httpx is an HTTP client for calling the Adzuna API.
- pydantic-settings loads your .env file into typed Python settings.
- Confirm the install succeeded by checking your pyproject.toml:
cat pyproject.tomlYou should see all five packages listed under dependencies with their version constraints.
- In pyproject.toml, find this line:
description = "Add your description here"- Replace it with:
description = "Job Aggregator API with FastAPI and Adzuna"Edit with your terminal or editor
You can open pyproject.toml in any text editor to make this change. You will open VS Code in the next substep, so you can also wait and edit it there.
Seeing a resolution error?
If uv add fails with a resolution error, make sure your Python version is 3.11 or higher. Run python3 --version to confirm.
help me fix a uv add resolution error for my FastAPI project
Register for a free Adzuna API key
Adzuna is a job search engine that provides a free API with real job listing data. You need an API key to authenticate your requests.
- Open your browser and navigate to https://developer.adzuna.com.
- Click Sign Up and create a free account.
- After registration, you will receive an app_id and an app_key. Note both of these values somewhere safe.
What is the Adzuna free tier?
The Adzuna free tier allows 2,500 requests per month. That is more than enough for building and testing this project.
- Back in your terminal, create a .env file in your project directory by running:
touch .env- Confirm the file exists by running:
ls -aYou should see .env listed among the files.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type VS Code and press Enter to open it.
- In VS Code, click File in the top menu bar.
- Click Open Folder.
- Navigate to your Desktop, select the job-aggregator-api folder, then click Open.
- In the Explorer sidebar on the left, click on .env to open it.
- Paste the following into .env:
ADZUNA_APP_ID=your-app-id-here
ADZUNA_APP_KEY=your-app-key-here
DATABASE_URL=postgresql://localhost/job_aggregator- Replace your-app-id-here with the app_id from your Adzuna developer dashboard.
- Replace your-app-key-here with the app_key from your Adzuna developer dashboard.
- Save the file.
What is DATABASE_URL?
DATABASE_URL tells your app how to connect to PostgreSQL. The value postgresql://localhost/job_aggregator points to a database called job_aggregator on your local machine. You will create this database in the next step.
Not sure where to find your Adzuna credentials?
After signing up at developer.adzuna.com, your app_id and app_key are displayed on your dashboard page. If you cannot find them, check your registration confirmation email.
help me find my Adzuna API credentials
Create the project file structure and verify the server
FastAPI projects use a modular file structure. You will create the app/ directory with separate files for configuration, models, schemas, routes, and services.
- In VS Code, open the integrated terminal by pressing Ctrl+` (backtick key, same on macOS and Windows).
- Create the full directory structure by running these commands:
mkdir -p app/routes app/services
touch app/__init__.py app/main.py app/config.py app/models.py app/schemas.py app/database.py
touch app/routes/__init__.py app/services/__init__.pyWhat does this structure do?
- app/main.py is the entry point where FastAPI starts.
- app/config.py loads settings from your .env file.
- app/models.py will define your database tables.
- app/schemas.py will define request and response shapes.
- app/database.py will handle the database connection.
- app/routes/ holds your API endpoint definitions.
- app/services/ holds business logic like the Adzuna integration.
- Confirm the files exist by running:
ls app/ app/routes/ app/services/You should see __init__.py, main.py, config.py, models.py, schemas.py, and database.py under app/. Both routes/ and services/ should each contain __init__.py.
- In the Explorer sidebar, click on app/config.py to open it.
- Paste the following code into app/config.py:
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
adzuna_app_id: str
adzuna_app_key: str
database_url: str
model_config = {"env_file": ".env"}
settings = Settings()What does this code do?
- Settings is a Pydantic class that automatically reads your .env file and validates that all required values are present.
- model_config = {"env_file": ".env"} tells Pydantic where to find the environment variables.
- settings = Settings() creates a single instance you can import anywhere in your app.
- Save app/config.py.
- In the Explorer sidebar, click on app/main.py to open it.
- Paste the following code into app/main.py:
from fastapi import FastAPI
# Create the FastAPI application instance with metadata for docs
app = FastAPI(
title="Job Aggregator API",
description="Aggregates job listings from Adzuna with comprehensive filtering",
version="0.1.0",
)
@app.get("/health")
def health_check():
return {"status": "ok"}What does this code do?
- FastAPI() creates the application instance with metadata that appears in the auto-generated Swagger documentation.
- @app.get("/health") defines a simple endpoint at /health that returns a JSON status. This lets you quickly verify the server is running.
- Save app/main.py.
- Start the FastAPI development server by running this command in your terminal:
uv run fastapi dev app/main.pyWhat does this command do?
uv run executes a command inside the project's virtual environment. fastapi dev starts the development server with hot-reloading enabled, so code changes take effect immediately without restarting.
- Open your browser and navigate to http://127.0.0.1:8000/docs.
You should see the Swagger UI page with the title "Job Aggregator API" and a single /health endpoint listed.
Server not starting?
Make sure your .env file has values for ADZUNA_APP_ID and ADZUNA_APP_KEY. Even placeholder text like your-app-id-here works for now. The config.py file only checks that the values exist, not that they are valid API credentials.
Also confirm you are in the job-aggregator-api directory when running the command. Run pwd to check your current location.
help me debug why my FastAPI dev server is not starting
✔️ Awesome, I've got everything!
Great. Double check you have saved both app/config.py and app/main.py.
ⓧ I'd like to double check the full code
Here is the complete code for each file at this point in the project.
[project]
name = "job-aggregator-api"
version = "0.1.0"
description = "Job Aggregator API with FastAPI and Adzuna"
requires-python = ">=3.11"
dependencies = [
"fastapi[standard]>=0.136.0,<0.137.0",
"sqlalchemy>=2.0.50,<2.1.0",
"psycopg2-binary>=2.9.12,<3.0.0",
"httpx>=0.28.1,<0.29.0",
"pydantic-settings>=2.0.0,<3.0.0",
]ADZUNA_APP_ID=your-app-id-here
ADZUNA_APP_KEY=your-app-key-here
DATABASE_URL=postgresql://localhost/job_aggregatorfrom pydantic_settings import BaseSettings
class Settings(BaseSettings):
adzuna_app_id: str
adzuna_app_key: str
database_url: str
model_config = {"env_file": ".env"}
settings = Settings()from fastapi import FastAPI
app = FastAPI(
title="Job Aggregator API",
description="Aggregates job listings from Adzuna with comprehensive filtering",
version="0.1.0",
)
@app.get("/health")
def health_check():
return {"status": "ok"}Your project foundation is in place. Next up, you will define the database models and schema that store the job listings your API fetches from Adzuna.
Define the Database Models and Schema
Your project is set up with all its dependencies installed and the dev server running.
Now you need a place to store job listings. In this step, you'll create a PostgreSQL database, define a SQLAlchemy model that maps Adzuna's job data to your own table, and build Pydantic schemas that control how data enters and leaves your API.
In this step, get ready to:
- Create a PostgreSQL database and configure SQLAlchemy to connect to it.
- Define the Job model with fields for everything Adzuna returns.
- Build Pydantic response schemas and a filter dependency for query parameters.
Create the database and configure SQLAlchemy
Before your API can store jobs, you need an actual PostgreSQL database for it to connect to.
- Create the job_aggregator database by running this command in your terminal:
createdb job_aggregator- Confirm the database exists by running:
psql -d job_aggregator -c "\dt"You should see a message like "Did not find any relations." because the database is empty for now.
Seeing 'database does not exist'?
Make sure PostgreSQL is running on your machine. On macOS, check that the Postgres app is active or run brew services start postgresql if you installed via Homebrew.
help me troubleshoot my PostgreSQL createdb command failing
Now you need to tell your application how to connect to this database. The DATABASE_URL in your .env file already points to postgresql://localhost/job_aggregator. SQLAlchemy uses this to open a connection pool.
- In VS Code, open app/database.py and add the following code:
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker
from app.config import settings
# Create the database engine using the connection URL from .env
engine = create_engine(settings.database_url)
# Create a session factory bound to the engine
SessionLocal = sessionmaker(bind=engine)
# Base class that all models will inherit from
class Base(DeclarativeBase):
pass
# Dependency that provides a database session to each request
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()What does this code do?
This file sets up the connection between your Python code and PostgreSQL.
- engine is a connection pool that manages how your app talks to the database.
- SessionLocal is a factory that creates new database sessions (one per request).
- Base is the parent class every model inherits from. SQLAlchemy uses it to track which tables exist.
- get_db() is a FastAPI dependency that opens a session, lets the route use it, then closes it automatically.
- Save app/database.py.
You won't see a change yet. The next chunk uses this module to define your Job model.
Define the Job model
The Job model maps every field that Adzuna returns to a column in your database. This is where you decide what data to keep and how to index it for fast filtering later.
- Open app/models.py and add the Job model:
from datetime import datetime
from sqlalchemy import (
Boolean,
DateTime,
Float,
Index,
Integer,
JSON,
String,
Text,
ForeignKey,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
class Job(Base):
__tablename__ = "jobs"
# Primary key and unique external reference from Adzuna
id: Mapped[int] = mapped_column(Integer, primary_key=True)
external_id: Mapped[str] = mapped_column(String(50), unique=True, index=True)
# Core job details
title: Mapped[str] = mapped_column(String(500))
company: Mapped[str | None] = mapped_column(String(300))
description: Mapped[str | None] = mapped_column(Text)
# Location data (display_name as string, area as JSON array)
location_display: Mapped[str | None] = mapped_column(String(300))
location_area: Mapped[list | None] = mapped_column(JSON)
# Category for filtering by job type
category_label: Mapped[str | None] = mapped_column(String(200))
category_tag: Mapped[str | None] = mapped_column(String(100), index=True)
# Salary range for range-based filtering
salary_min: Mapped[float | None] = mapped_column(Float, index=True)
salary_max: Mapped[float | None] = mapped_column(Float)
salary_is_predicted: Mapped[bool] = mapped_column(Boolean, default=False)
# Contract details
contract_type: Mapped[str | None] = mapped_column(String(50))
contract_time: Mapped[str | None] = mapped_column(String(50))
# Link back to the original posting
redirect_url: Mapped[str | None] = mapped_column(Text)
# Timestamps: when Adzuna created it vs when we fetched it
created_at: Mapped[datetime | None] = mapped_column(DateTime)
fetched_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
# Additional index for location-based queries
__table_args__ = (
Index("ix_jobs_location", "location_display"),
)What does this model do?
Each field maps to a column in the jobs table. The key design decisions are:
- external_id stores Adzuna's job ID so you can avoid inserting duplicates when syncing.
- location_area uses a JSON column because Adzuna returns location as a nested array (e.g. ["US", "California", "San Francisco"]).
- Indexes on category_tag, salary_min, and location_display speed up the filters you'll build in Step 4.
- Save app/models.py.
Now you need to tell app/main.py to create the tables when the app starts. This is a small edit to wire everything together.
- Open app/main.py and find this line at the top:
from fastapi import FastAPI- Replace the top section of the file so it looks like this:
from fastapi import FastAPI
from app.database import Base, engine
# Import models so SQLAlchemy registers them with Base
import app.models # noqa: F401
# Create all tables defined by models that inherit from Base
Base.metadata.create_all(bind=engine)
app = FastAPI(
title="Job Aggregator API",
description="Aggregates job listings from Adzuna with comprehensive filtering",
version="0.1.0",
)
@app.get("/health")
def health_check():
return {"status": "ok"}Why import app.models here?
SQLAlchemy only knows about tables defined in classes that inherit from Base. Importing app.models ensures the Job class is registered before create_all runs. Without it, no tables get created.
- Save app/main.py.
- Restart your FastAPI dev server by running:
uv run fastapi dev app/main.py- In a separate terminal window, verify the table was created by running:
psql -d job_aggregator -c "\dt"You should see a table named jobs listed under the public schema.
Don't see the jobs table?
Check that your DATABASE_URL in .env matches the database you created (postgresql://localhost/job_aggregator).
Make sure you saved both app/models.py and app/main.py before restarting the server. SQLAlchemy needs the model imported before create_all runs.
help me debug why my SQLAlchemy tables are not being created in PostgreSQL
Define the response schemas and filter parameters
Pydantic schemas control the shape of data your API sends back to clients. They also validate incoming query parameters so your filtering logic receives clean, typed values.
- Open app/schemas.py and add the response models:
from datetime import datetime
from fastapi import Query
from pydantic import BaseModel
# Response model for a single job listing
class JobResponse(BaseModel):
id: int
external_id: str
title: str
company: str | None
description: str | None
location_display: str | None
location_area: list | None
category_label: str | None
category_tag: str | None
salary_min: float | None
salary_max: float | None
salary_is_predicted: bool
contract_type: str | None
contract_time: str | None
redirect_url: str | None
created_at: datetime | None
fetched_at: datetime
model_config = {"from_attributes": True}
# Metadata for paginated responses
class PaginationMeta(BaseModel):
page: int
page_size: int
total: int
total_pages: int
# Wraps a list of jobs with pagination info
class JobListResponse(BaseModel):
jobs: list[JobResponse]
meta: PaginationMeta
# Request body for triggering a job sync from Adzuna
class SyncRequest(BaseModel):
country: str = "us"
what: str = ""
where: str = ""
pages: int = 1
# Response after a sync completes
class SyncResponse(BaseModel):
message: str
jobs_synced: intWhat do these models do?
- JobResponse defines what fields a client sees when requesting a job. The from_attributes = True setting lets Pydantic read directly from SQLAlchemy model instances.
- PaginationMeta gives the client everything it needs to build pagination controls (current page, total results, total pages).
- SyncRequest and SyncResponse are for the endpoint that pulls jobs from Adzuna into your database.
Next, add the filter dependency class below the SyncResponse class in the same file.
- Add the JobFilterParams class at the bottom of app/schemas.py:
# Dependency class that FastAPI injects with validated query parameters
class JobFilterParams:
def __init__(
self,
search: str | None = Query(None, description="Search in title and description"),
category: str | None = Query(None, description="Filter by category tag"),
location: str | None = Query(None, description="Filter by location"),
salary_min: float | None = Query(None, description="Minimum salary"),
salary_max: float | None = Query(None, description="Maximum salary"),
contract_type: str | None = Query(None, description="e.g. permanent, contract"),
contract_time: str | None = Query(None, description="e.g. full_time, part_time"),
sort_by: str = Query("date", description="Sort: date, salary_asc, salary_desc"),
page: int = Query(1, ge=1, description="Page number"),
page_size: int = Query(20, ge=1, le=100, description="Results per page"),
):
self.search = search
self.category = category
self.location = location
self.salary_min = salary_min
self.salary_max = salary_max
self.contract_type = contract_type
self.contract_time = contract_time
self.sort_by = sort_by
self.page = page
self.page_size = page_sizeWhat is a dependency class?
JobFilterParams is not a Pydantic model. It is a plain Python class that FastAPI can inject as a dependency using Depends(). FastAPI reads the __init__ signature and automatically parses query parameters from the URL.
Each Query() call defines a default value, validation constraints (like ge=1 for minimum page number), and a description that shows up in Swagger UI.
- Save app/schemas.py.
- Confirm the server is still running without errors in your terminal. You should see no tracebacks.
Seeing an import or syntax error?
Make sure the from fastapi import Query line is at the top of app/schemas.py. Without it, the Query() calls inside JobFilterParams will fail.
Also check that you have no extra indentation on the class. The class JobFilterParams: line should be at the module level, not nested inside another class.
help me fix my schemas.py syntax error
✔️ Awesome, I've got everything!
Great. Double check that you've saved all three files: app/database.py, app/models.py, and app/schemas.py.
ⓧ I'd like to double check the full code
Here is the full code for each file at this point:
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker
from app.config import settings
engine = create_engine(settings.database_url)
SessionLocal = sessionmaker(bind=engine)
class Base(DeclarativeBase):
pass
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()from datetime import datetime
from sqlalchemy import (
Boolean,
DateTime,
Float,
Index,
Integer,
JSON,
String,
Text,
ForeignKey,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
class Job(Base):
__tablename__ = "jobs"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
external_id: Mapped[str] = mapped_column(String(50), unique=True, index=True)
title: Mapped[str] = mapped_column(String(500))
company: Mapped[str | None] = mapped_column(String(300))
description: Mapped[str | None] = mapped_column(Text)
location_display: Mapped[str | None] = mapped_column(String(300))
location_area: Mapped[list | None] = mapped_column(JSON)
category_label: Mapped[str | None] = mapped_column(String(200))
category_tag: Mapped[str | None] = mapped_column(String(100), index=True)
salary_min: Mapped[float | None] = mapped_column(Float, index=True)
salary_max: Mapped[float | None] = mapped_column(Float)
salary_is_predicted: Mapped[bool] = mapped_column(Boolean, default=False)
contract_type: Mapped[str | None] = mapped_column(String(50))
contract_time: Mapped[str | None] = mapped_column(String(50))
redirect_url: Mapped[str | None] = mapped_column(Text)
created_at: Mapped[datetime | None] = mapped_column(DateTime)
fetched_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
__table_args__ = (
Index("ix_jobs_location", "location_display"),
)from datetime import datetime
from fastapi import Query
from pydantic import BaseModel
class JobResponse(BaseModel):
id: int
external_id: str
title: str
company: str | None
description: str | None
location_display: str | None
location_area: list | None
category_label: str | None
category_tag: str | None
salary_min: float | None
salary_max: float | None
salary_is_predicted: bool
contract_type: str | None
contract_time: str | None
redirect_url: str | None
created_at: datetime | None
fetched_at: datetime
model_config = {"from_attributes": True}
class PaginationMeta(BaseModel):
page: int
page_size: int
total: int
total_pages: int
class JobListResponse(BaseModel):
jobs: list[JobResponse]
meta: PaginationMeta
class SyncRequest(BaseModel):
country: str = "us"
what: str = ""
where: str = ""
pages: int = 1
class SyncResponse(BaseModel):
message: str
jobs_synced: int
class JobFilterParams:
def __init__(
self,
search: str | None = Query(None, description="Search in title and description"),
category: str | None = Query(None, description="Filter by category tag"),
location: str | None = Query(None, description="Filter by location"),
salary_min: float | None = Query(None, description="Minimum salary"),
salary_max: float | None = Query(None, description="Maximum salary"),
contract_type: str | None = Query(None, description="e.g. permanent, contract"),
contract_time: str | None = Query(None, description="e.g. full_time, part_time"),
sort_by: str = Query("date", description="Sort: date, salary_asc, salary_desc"),
page: int = Query(1, ge=1, description="Page number"),
page_size: int = Query(20, ge=1, le=100, description="Results per page"),
):
self.search = search
self.category = category
self.location = location
self.salary_min = salary_min
self.salary_max = salary_max
self.contract_type = contract_type
self.contract_time = contract_time
self.sort_by = sort_by
self.page = page
self.page_size = page_sizefrom fastapi import FastAPI
from app.database import Base, engine
Base.metadata.create_all(bind=engine)
app = FastAPI(
title="Job Aggregator API",
description="Aggregates job listings from Adzuna with comprehensive filtering",
version="0.1.0",
)
@app.get("/health")
def health_check():
return {"status": "ok"}Your database is structured and ready to receive job listings. Next up, you'll build the service that fetches real jobs from Adzuna and fills this table with data.
Build the Adzuna Integration Service
Your database schema is ready and waiting for data. The next step is to actually fill it with real job listings from an external source.
In this step, you will build a service layer that calls the Adzuna API, normalizes the response data into your Job model, and stores everything in PostgreSQL. You will also create an endpoint that triggers this sync on demand.
In this step, get ready to:
- Create an AdzunaService class that fetches job listings from the Adzuna search API.
- Implement a sync method that maps Adzuna's nested response fields to your database model.
- Wire up a POST endpoint to trigger the sync and populate your database.
Create the AdzunaService class with a fetch method
The Adzuna API returns job listings as JSON with nested objects for location, company, and category. Your service class will handle the HTTP calls and give the rest of your app a clean interface to work with.
- Create a new file called adzuna.py inside the app/services/ directory.
- Add the following code to define the AdzunaService class with its fetch_jobs method:
from datetime import datetime
import httpx
from sqlalchemy.orm import Session
from app.config import settings
from app.models import Job
class AdzunaService:
# Base URL for all Adzuna API calls
BASE_URL = "https://api.adzuna.com/v1/api"
def __init__(self):
# Load credentials from environment settings
self.app_id = settings.adzuna_app_id
self.app_key = settings.adzuna_app_key
def fetch_jobs(
self,
country: str = "us",
what: str = "",
where: str = "",
page: int = 1,
results_per_page: int = 50,
) -> list[dict]:
# Build the search endpoint URL with country and page number
url = f"{self.BASE_URL}/jobs/{country}/search/{page}"
params = {
"app_id": self.app_id,
"app_key": self.app_key,
"results_per_page": results_per_page,
"content-type": "application/json",
}
# Only include search filters if provided
if what:
params["what"] = what
if where:
params["where"] = where
# Make the HTTP GET request to Adzuna
response = httpx.get(url, params=params, timeout=30.0)
response.raise_for_status()
data = response.json()
return data.get("results", [])What does this code do?
The AdzunaService class wraps all communication with the Adzuna API. It reads your app_id and app_key from the settings you configured in .env.
The fetch_jobs method builds the Adzuna search URL using a country code and page number, attaches your credentials as query parameters, and returns the raw list of job result dictionaries.
Why a service class instead of inline API calls?
Separating the API integration into its own service keeps your route handlers thin. The route only knows "call sync_jobs" without caring about HTTP details, pagination, or field mapping.
This mirrors production patterns where data ingestion runs independently from the read API. You could swap Adzuna for a different provider later without touching your endpoints.
- Save app/services/adzuna.py.
You will not see a visible change yet. The next chunk adds the method that actually writes jobs to the database.
Add the sync_jobs method to store fetched jobs
Adzuna's response nests company, location, and category inside child objects. The sync_jobs method will extract those nested fields, check for duplicates using external_id, and insert new records into PostgreSQL.
- In app/services/adzuna.py, add the following sync_jobs method inside the AdzunaService class, below the fetch_jobs method:
def sync_jobs(
self,
db: Session,
country: str = "us",
what: str = "",
where: str = "",
pages: int = 1,
) -> int:
total_synced = 0
for page in range(1, pages + 1):
# Fetch one page of results from Adzuna
results = self.fetch_jobs(
country=country, what=what, where=where, page=page
)
for item in results:
external_id = str(item.get("id", ""))
if not external_id:
continue
# Skip jobs we already have in the database
existing = (
db.query(Job).filter(Job.external_id == external_id).first()
)
if existing:
continue
# Parse the ISO timestamp from Adzuna
created_at = None
if item.get("created"):
try:
created_at = datetime.fromisoformat(
item["created"].replace("Z", "+00:00")
)
except (ValueError, TypeError):
pass
# Map nested Adzuna fields to flat Job model columns
job = Job(
external_id=external_id,
title=item.get("title", ""),
company=item.get("company", {}).get("display_name"),
description=item.get("description"),
location_display=item.get("location", {}).get("display_name"),
location_area=item.get("location", {}).get("area"),
category_label=item.get("category", {}).get("label"),
category_tag=item.get("category", {}).get("tag"),
salary_min=item.get("salary_min"),
salary_max=item.get("salary_max"),
salary_is_predicted=bool(item.get("salary_is_predicted", 0)),
contract_type=item.get("contract_type"),
contract_time=item.get("contract_time"),
redirect_url=item.get("redirect_url"),
created_at=created_at,
)
db.add(job)
total_synced += 1
# Commit after each page to avoid holding a large transaction
db.commit()
return total_syncedWhat does this code do?
The sync_jobs method loops through the requested number of pages from Adzuna. For each job result, it checks whether a record with that external_id already exists to prevent duplicates.
The field mapping extracts nested values like company.display_name, location.display_name, and category.tag into flat columns on your Job model. It commits after each page so a failure on page 3 does not lose pages 1 and 2.
- Save app/services/adzuna.py.
You still will not see a visible change. The next chunk creates the endpoint that calls this method.
✔️ Awesome, I've got everything!
Great. Double check you have saved app/services/adzuna.py with both methods.
ⓧ I'd like to double check the full code
from datetime import datetime
import httpx
from sqlalchemy.orm import Session
from app.config import settings
from app.models import Job
class AdzunaService:
BASE_URL = "https://api.adzuna.com/v1/api"
def __init__(self):
self.app_id = settings.adzuna_app_id
self.app_key = settings.adzuna_app_key
def fetch_jobs(
self,
country: str = "us",
what: str = "",
where: str = "",
page: int = 1,
results_per_page: int = 50,
) -> list[dict]:
url = f"{self.BASE_URL}/jobs/{country}/search/{page}"
params = {
"app_id": self.app_id,
"app_key": self.app_key,
"results_per_page": results_per_page,
"content-type": "application/json",
}
if what:
params["what"] = what
if where:
params["where"] = where
response = httpx.get(url, params=params, timeout=30.0)
response.raise_for_status()
data = response.json()
return data.get("results", [])
def sync_jobs(
self,
db: Session,
country: str = "us",
what: str = "",
where: str = "",
pages: int = 1,
) -> int:
total_synced = 0
for page in range(1, pages + 1):
results = self.fetch_jobs(
country=country, what=what, where=where, page=page
)
for item in results:
external_id = str(item.get("id", ""))
if not external_id:
continue
existing = (
db.query(Job).filter(Job.external_id == external_id).first()
)
if existing:
continue
created_at = None
if item.get("created"):
try:
created_at = datetime.fromisoformat(
item["created"].replace("Z", "+00:00")
)
except (ValueError, TypeError):
pass
job = Job(
external_id=external_id,
title=item.get("title", ""),
company=item.get("company", {}).get("display_name"),
description=item.get("description"),
location_display=item.get("location", {}).get("display_name"),
location_area=item.get("location", {}).get("area"),
category_label=item.get("category", {}).get("label"),
category_tag=item.get("category", {}).get("tag"),
salary_min=item.get("salary_min"),
salary_max=item.get("salary_max"),
salary_is_predicted=bool(item.get("salary_is_predicted", 0)),
contract_type=item.get("contract_type"),
contract_time=item.get("contract_time"),
redirect_url=item.get("redirect_url"),
created_at=created_at,
)
db.add(job)
total_synced += 1
db.commit()
return total_syncedCreate the sync endpoint and wire up the router
Your service is ready but has no way to be triggered. You need a POST endpoint that accepts a search request and calls sync_jobs to populate the database.
- Create a new file called jobs.py inside the app/routes/ directory.
- Add the following code to define the sync endpoint:
import math
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import Job
from app.schemas import (
SyncRequest,
SyncResponse,
)
from app.services.adzuna import AdzunaService
# Create a router to group job-related endpoints
router = APIRouter()
adzuna_service = AdzunaService()
@router.post("/jobs/sync", response_model=SyncResponse)
def sync_jobs(request: SyncRequest, db: Session = Depends(get_db)):
# Trigger the Adzuna sync with the request parameters
count = adzuna_service.sync_jobs(
db=db,
country=request.country,
what=request.what,
where=request.where,
pages=request.pages,
)
return SyncResponse(message="Sync complete", jobs_synced=count)What does this code do?
The APIRouter groups all job-related endpoints together. The POST /jobs/sync endpoint accepts a SyncRequest body (country, keyword, location, number of pages) and passes it to the AdzunaService.
The Depends(get_db) injection gives the endpoint a database session that is automatically closed after the request finishes. The response tells the caller how many new jobs were stored.
- Save app/routes/jobs.py.
Now you need to register this router with your FastAPI application so the endpoint becomes accessible.
- In app/main.py, find this line near the top of the file:
from app.database import Base, engine- Replace it with the following two imports:
from app.database import Base, engine
from app.routes.jobs import router- Then find this block:
app = FastAPI(
title="Job Aggregator API",
description="Aggregates job listings from Adzuna with comprehensive filtering",
version="0.1.0",
)- Add the following line directly below it:
app.include_router(router)What does include_router do?
The include_router call mounts all routes defined in app/routes/jobs.py onto the main FastAPI app. Without this line, FastAPI would not know about your /jobs/sync endpoint.
- Save app/main.py.
- Your dev server should automatically reload. Open http://127.0.0.1:8000/docs in your browser.
- You should see a new POST /jobs/sync endpoint listed in the Swagger UI.
Don't see the new endpoint?
Check your terminal for import errors. Make sure app/services/__init__.py and app/routes/__init__.py both exist (they can be empty files).
Confirm that the from app.routes.jobs import router line is at the top of app/main.py, not inside a function.
Help me debug why my FastAPI server is not loading the jobs router.
Now test the sync by clicking POST /jobs/sync in Swagger UI, then clicking Try it out.
- Leave the request body as the defaults (country us, empty what and where, 1 page) and click Execute.
- You should see a response like {"message": "Sync complete", "jobs_synced": 50}.
- Confirm jobs are in your database by running this command in a separate terminal:
psql -d job_aggregator -c "SELECT COUNT(*) FROM jobs;"You should see a count of 50 (or close to it, depending on Adzuna's current results).
Seeing a 401 or 403 error?
This means your Adzuna credentials are incorrect. Double check that ADZUNA_APP_ID and ADZUNA_APP_KEY in your .env file match exactly what the Adzuna developer dashboard shows. There should be no extra spaces or quotes around the values.
Help me fix my Adzuna API authentication error.
✔️ Awesome, I've got everything!
Great. Double check you have saved all three files: app/services/adzuna.py, app/routes/jobs.py, and app/main.py.
ⓧ I'd like to double check the full code
import math
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import Job
from app.schemas import (
SyncRequest,
SyncResponse,
)
from app.services.adzuna import AdzunaService
router = APIRouter()
adzuna_service = AdzunaService()
@router.post("/jobs/sync", response_model=SyncResponse)
def sync_jobs(request: SyncRequest, db: Session = Depends(get_db)):
count = adzuna_service.sync_jobs(
db=db,
country=request.country,
what=request.what,
where=request.where,
pages=request.pages,
)
return SyncResponse(message="Sync complete", jobs_synced=count)from fastapi import FastAPI
from app.database import Base, engine
from app.routes.jobs import router
Base.metadata.create_all(bind=engine)
app = FastAPI(
title="Job Aggregator API",
description="Aggregates job listings from Adzuna with comprehensive filtering",
version="0.1.0",
)
app.include_router(router)
@app.get("/health")
def health_check():
return {"status": "ok"}Your API can now pull real job data from Adzuna into your local database on demand. Next up, you will build the filtering and search endpoints that let clients query this data with precision.
Build the Filtering and Search Endpoints
Your database has job listings stored from Adzuna, but there is no way for a client (like your future Flutter app) to search or filter through them.
In this step, you will build the read endpoints that make your data useful. A mobile app needs to search by keyword, narrow results by salary, filter by contract type, and page through results without loading everything at once.
In this step, get ready to:
- Create a GET /jobs endpoint with dynamic query filtering.
- Add pagination and sorting to the job list response.
- Create a GET /jobs/{job_id} endpoint for single job details.
Create the GET /jobs endpoint with dynamic filtering
A real job search needs to combine multiple filters at once. Someone might want Python jobs in New York paying over $100k. Your endpoint will build a SQLAlchemy query dynamically, applying only the filters the client provides.
- In app/routes/jobs.py, find the import section at the top of the file and replace it with the following:
import math
from fastapi import APIRouter, Depends
from sqlalchemy import or_
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import Job
from app.schemas import (
JobFilterParams,
JobListResponse,
JobResponse,
PaginationMeta,
SyncRequest,
SyncResponse,
)
from app.services.adzuna import AdzunaServiceWhat changed in the imports?
You added or_ from SQLAlchemy, which lets you combine multiple filter conditions with OR logic (e.g., search in title OR description).
You also imported the new schema classes: JobFilterParams (accepts query parameters), JobListResponse (wraps the response), JobResponse (single job shape), and PaginationMeta (page metadata).
- Below the existing sync_jobs endpoint, add the list_jobs endpoint by pasting this code:
@router.get("/jobs", response_model=JobListResponse)
def list_jobs(
filters: JobFilterParams = Depends(),
db: Session = Depends(get_db),
):
query = db.query(Job)
# Keyword search across title and description
if filters.search:
search_term = f"%{filters.search}%"
query = query.filter(
or_(
Job.title.ilike(search_term),
Job.description.ilike(search_term),
)
)
# Exact match filters
if filters.category:
query = query.filter(Job.category_tag == filters.category)
# Partial match on location
if filters.location:
query = query.filter(Job.location_display.ilike(f"%{filters.location}%"))
# Salary range filters
if filters.salary_min is not None:
query = query.filter(Job.salary_min >= filters.salary_min)
if filters.salary_max is not None:
query = query.filter(Job.salary_max <= filters.salary_max)
# Contract filters
if filters.contract_type:
query = query.filter(Job.contract_type == filters.contract_type)
if filters.contract_time:
query = query.filter(Job.contract_time == filters.contract_time)
# Get total count before pagination
total = query.count()
# Apply sorting
if filters.sort_by == "salary_asc":
query = query.order_by(Job.salary_min.asc().nulls_last())
elif filters.sort_by == "salary_desc":
query = query.order_by(Job.salary_min.desc().nulls_last())
else:
query = query.order_by(Job.created_at.desc().nulls_last())
# Apply pagination
offset = (filters.page - 1) * filters.page_size
jobs = query.offset(offset).limit(filters.page_size).all()
total_pages = math.ceil(total / filters.page_size) if total > 0 else 0
return JobListResponse(
jobs=jobs,
meta=PaginationMeta(
page=filters.page,
page_size=filters.page_size,
total=total,
total_pages=total_pages,
),
)What does this code do?
This endpoint builds a database query piece by piece, applying only the filters the client sends:
- ilike performs case-insensitive partial matching. It wraps the search term in % wildcards so "python" matches "Senior Python Developer".
- or_ combines title and description searches so a keyword hit in either field returns the job.
- Salary filters use is not None checks because 0 is a valid salary value that would be falsy in a bare if check.
- nulls_last() ensures jobs without salary data appear at the end of sorted results, not the top.
- Save app/routes/jobs.py.
- Your dev server should auto-reload. You will not see a visible change yet because we still need to add the single-job endpoint before testing.
Seeing an import error on reload?
- Check that from sqlalchemy import or_ is on its own line, separate from the sqlalchemy.orm import below it.
- Make sure JobFilterParams is included in the schemas import block. It is the class defined in app/schemas.py from the previous step.
Help me debug an import error in my jobs route file.
Understand how pagination and sorting work
A mobile app cannot load thousands of jobs at once. Pagination lets the client request one page at a time, and sorting controls which jobs appear first.
How does the pagination logic work?
The function counts the total matching rows BEFORE applying offset/limit. This gives the client a total count and total_pages value so it can render page controls.
The offset formula (page - 1) * page_size skips rows from previous pages. For example, page 3 with 20 results per page skips the first 40 rows and returns rows 41 through 60.
The default sort is by created_at descending (newest jobs first). Clients can override this by passing sort_by=salary_asc or sort_by=salary_desc as query parameters.
Create the GET /jobs/{job_id} endpoint and test
Clients also need to fetch the full details of a single job when a user taps on it in the app.
- In app/routes/jobs.py, add this endpoint below the list_jobs function:
@router.get("/jobs/{job_id}", response_model=JobResponse)
def get_job(job_id: int, db: Session = Depends(get_db)):
job = db.query(Job).filter(Job.id == job_id).first()
if not job:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Job not found")
return jobWhat does this code do?
- The path parameter {job_id} captures the ID from the URL. FastAPI automatically validates it as an integer.
- If no job matches the ID, the endpoint returns a 404 status code with a clear error message instead of crashing.
- The HTTPException import is inside the function to keep it local. It is only needed when the error path triggers.
- Save app/routes/jobs.py.
- Switch to your browser and open http://127.0.0.1:8000/docs.
- Expand the GET /jobs endpoint and click Try it out.
- Enter python in the search field.
- Enter 50000 in the salary_min field.
- Enter salary_desc in the sort_by field.
- Click Execute.
- You should see a JSON response with a jobs array and a meta object containing page, page_size, total, and total_pages.
Getting an empty jobs array?
- Make sure you synced jobs in the previous step using POST /jobs/sync. If your database is empty, no filters will return results.
- Try removing all filter values and querying GET /jobs with no parameters first to confirm jobs exist.
- If you see a 500 error, check your terminal for the traceback. A common cause is a typo in a column name like location_display vs location.
Help me debug why my GET /jobs endpoint returns no results.
Why separate sync from search?
In production APIs, data ingestion (writing) and data serving (reading) are separate concerns. Your POST /jobs/sync handles fetching from Adzuna, while GET /jobs only reads from your local database.
This means search responses are fast (no external API calls) and you can schedule syncs independently of user traffic.
✔️ Awesome, I've got everything!
Great. Double check you have saved app/routes/jobs.py with both the list_jobs and get_job endpoints.
ⓧ I'd like to double check the full code
Here is the complete app/routes/jobs.py file at the end of this step:
import math
from fastapi import APIRouter, Depends
from sqlalchemy import or_
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import Job
from app.schemas import (
JobFilterParams,
JobListResponse,
JobResponse,
PaginationMeta,
SyncRequest,
SyncResponse,
)
from app.services.adzuna import AdzunaService
router = APIRouter()
adzuna_service = AdzunaService()
@router.post("/jobs/sync", response_model=SyncResponse)
def sync_jobs(request: SyncRequest, db: Session = Depends(get_db)):
count = adzuna_service.sync_jobs(
db=db,
country=request.country,
what=request.what,
where=request.where,
pages=request.pages,
)
return SyncResponse(message="Sync complete", jobs_synced=count)
@router.get("/jobs", response_model=JobListResponse)
def list_jobs(
filters: JobFilterParams = Depends(),
db: Session = Depends(get_db),
):
query = db.query(Job)
if filters.search:
search_term = f"%{filters.search}%"
query = query.filter(
or_(
Job.title.ilike(search_term),
Job.description.ilike(search_term),
)
)
if filters.category:
query = query.filter(Job.category_tag == filters.category)
if filters.location:
query = query.filter(Job.location_display.ilike(f"%{filters.location}%"))
if filters.salary_min is not None:
query = query.filter(Job.salary_min >= filters.salary_min)
if filters.salary_max is not None:
query = query.filter(Job.salary_max <= filters.salary_max)
if filters.contract_type:
query = query.filter(Job.contract_type == filters.contract_type)
if filters.contract_time:
query = query.filter(Job.contract_time == filters.contract_time)
total = query.count()
if filters.sort_by == "salary_asc":
query = query.order_by(Job.salary_min.asc().nulls_last())
elif filters.sort_by == "salary_desc":
query = query.order_by(Job.salary_min.desc().nulls_last())
else:
query = query.order_by(Job.created_at.desc().nulls_last())
offset = (filters.page - 1) * filters.page_size
jobs = query.offset(offset).limit(filters.page_size).all()
total_pages = math.ceil(total / filters.page_size) if total > 0 else 0
return JobListResponse(
jobs=jobs,
meta=PaginationMeta(
page=filters.page,
page_size=filters.page_size,
total=total,
total_pages=total_pages,
),
)
@router.get("/jobs/{job_id}", response_model=JobResponse)
def get_job(job_id: int, db: Session = Depends(get_db)):
job = db.query(Job).filter(Job.id == job_id).first()
if not job:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Job not found")
return jobYour API now has a powerful search layer. Clients can combine keyword searches, salary ranges, location filters, and contract types in a single request. Next up, you will tackle the Secret Mission and extend your API even further.
Secret mission
Add a Categories Endpoint and Saved Jobs
Your API fetches and filters jobs. But a real job search app needs two more features: browsable categories so users can discover what's available, and the ability to save interesting listings for later. Time to build both.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally with a free-tier API, so there are no ongoing costs.
Resources you used:
- A local PostgreSQL database (job_aggregator) containing job listings, categories, and saved jobs.
- An Adzuna developer account with API credentials (free tier, no ongoing cost).
- A project folder (job-aggregator-api) on your local machine.
Keep everything running
No action needed. Choose this if you're still actively building on your API or plan to connect a Flutter frontend.
- Your job_aggregator database stays intact with all synced jobs.
- Your Adzuna developer account remains active. It's free with no ongoing cost.
- Restart the dev server anytime by running uv run fastapi dev app/main.py from your project folder.
Pause - I'll come back to this later
Stop the running server to free up your terminal. Your data persists in PostgreSQL until you're ready to continue.
- Switch to the terminal running the FastAPI dev server.
- Press Ctrl+C to stop the server.
- Your database and project files remain on disk. Resume later by running uv run fastapi dev app/main.py from your project folder.
Delete - I don't want to use this again
Remove all project resources and start fresh. This deletes your database and local files permanently.
- Switch to the terminal running the FastAPI dev server and press Ctrl+C to stop it.
- Drop the PostgreSQL database by running this command:
dropdb job_aggregator- Delete the project folder. On macOS/Linux, run:
cd ~/Desktop
rm -rf job-aggregator-api- On Windows, run:
cd ~\Desktop
rmdir /s /q job-aggregator-api- Your Adzuna developer account is free with no ongoing costs. There is no financial reason to delete it.
Nice Work!
Nice Work!
Nice work! You just built a production-ready REST API that aggregates real job listings from an external data source into your own database and serves them through clean, filterable endpoints.
You've learned how to:
- Built a FastAPI backend with uv as the package manager and PostgreSQL as the database, integrating the Adzuna job listings API to fetch and normalize real job data.
- Designed a comprehensive filtering system with search, category, location, salary range, contract type, sorting, and pagination ready for mobile consumption.
- Created auto-generated OpenAPI documentation via Swagger UI ready for a Flutter app or any HTTP client to consume.
- Secret Mission: Added a categories endpoint with database caching and a saved-jobs bookmarking system with full CRUD operations.
Ready to quiz yourself?