Build a Device Management API with FastAPI
Build a REST API that registers IoT devices and ingests telemetry data.
Introduction
30 Second Summary
Every smart device you own, from temperature sensors to security cameras to fitness trackers, sends data back to a central server. That server needs to register each device, accept its readings, and keep a history of what happened.
In this project, you will build that server from scratch using Python and FastAPI. You will create a REST API that registers IoT devices, ingests telemetry readings, and stores everything in a database you can query through your browser.
What You'll Build
When you finish, you will open your browser, register a simulated ESP32-S3 device through interactive API docs, send it telemetry readings, and query its full history.
By the end of this project, you'll have:
- A working device registration API where you type a device name into the browser's Swagger UI, hit execute, and see it saved with a unique ID, type, and firmware version.
- A telemetry ingestion pipeline that accepts CPU usage, memory, Wi-Fi signal strength, and uptime readings from any registered device, then lets you query the full reading history in order.
- A persistent database layer backed by SQLite and abstracted through SQLAlchemy. Your data survives server restarts, and switching to PostgreSQL later is a one-line config change.
- Secret Mission: Add a device health endpoint that calculates online/offline status from the most recent telemetry timestamp.
Are there any prerequisites?
No prior API or database experience is needed. This guide walks you through installing Python, setting up a virtual environment, and building everything from a blank folder.
Before We Start
Before writing any code, let's lock in what you're about to build. You are creating the backend server for AURA OS, a REST API that registers edge devices, accepts telemetry readings, and lets you demo every endpoint live in your browser.
Set Up Python and the Project Structure
Every AURA OS component connects to a backend server. Before writing any server code, you need a working Python installation and a clean project scaffold.
The project structure you create in this step separates your code into distinct layers. Routes, database models, validation schemas, and business logic each get their own directory, mirroring how production Python backends are organized.
In this step, get ready to:
- Install Python 3.13 and verify it in the terminal.
- Create the aura-os-backend project with its modular directory structure.
- Set up a virtual environment and install project dependencies.
Install Python 3.13
Python is the language your entire backend runs on. You need version 3.13 or higher installed before anything else.
- In VS Code, open a new terminal by selecting Terminal > New Terminal from the top menu.
- Check if Python is already installed by running this command:
macOS
python3 --version
Windows
python --version
✔️ I see Python 3.13 or higher
Python is ready to go. Skip ahead to the next substep.
ⓧ I see an older version
You have Python installed, but this project requires 3.13 or higher. You need to install the correct version alongside your existing one.
macOS
macOS often ships with an older Python. The python.org installer adds the new version without removing your existing one.
- Download the macOS installer from python.org.
- Open the downloaded python-3.13.15-macos11.pkg file and follow the installer prompts.
- After installation, close and reopen the VS Code terminal, then run python3 --version again to confirm it now shows Python 3.13.15.
Why a separate install?
macOS bundles an older Python that some system tools depend on. The python.org installer places 3.13 in a separate location so both versions coexist safely. Use python3 (not python) to target the new version.
Windows
- Download the Windows installer (64-bit) from python.org.
- Run the installer. On the first screen, check the box that says Add python.exe to PATH before clicking Install Now.
- After installation, close and reopen the VS Code terminal, then run python --version again to confirm it now shows Python 3.13.15.
ⓧ Command not found
Python is not installed yet. Follow the instructions for your operating system to install it.
macOS
- Download the macOS installer from python.org.
- Open the downloaded python-3.13.15-macos11.pkg file and follow the installer prompts.
- After installation, close and reopen the VS Code terminal, then run python3 --version. You should see Python 3.13.15.
Windows
- Download the Windows installer (64-bit) from python.org.
- Run the installer. On the first screen, check the box that says Add python.exe to PATH before clicking Install Now.
- After installation, close and reopen the VS Code terminal, then run python --version. You should see Python 3.13.15.
Still seeing the old version after installing?
Make sure you closed and reopened the VS Code terminal after the install. The terminal caches the old PATH and will not find the new Python until it restarts.
On Windows, confirm you checked Add python.exe to PATH during install. If you missed it, re-run the installer and select Modify to add it.
help me troubleshoot my Python installation
Create the project folder and directory structure
Your backend will have six internal packages, each responsible for one layer of the application. Separating these from the start means each layer can grow without tangling into the others.
- Move to your Desktop by running this command:
cd ~/Desktop
- Create the project root folder and move into it by running these commands:
mkdir aura-os-backend
cd aura-os-backend
- Create all six package directories inside the app folder by running this command:
macOS
mkdir -p app/api app/models app/schemas app/services app/database app/core
Windows
mkdir app\api, app\models, app\schemas, app\services, app\database, app\core
- Create the empty __init__.py files that turn each directory into a Python package:
macOS
touch app/__init__.py app/api/__init__.py app/core/__init__.py app/database/__init__.py app/models/__init__.py app/schemas/__init__.py app/services/__init__.py
Windows
New-Item -ItemType File -Path app\__init__.py, app\api\__init__.py, app\core\__init__.py, app\database\__init__.py, app\models\__init__.py, app\schemas\__init__.py, app\services\__init__.py
What is __init__.py?
An __init__.py file tells Python that a directory is a package, not just a regular folder. Without it, you cannot import code from that directory.
- Confirm the folder structure was created correctly by running:
macOS
find app -type f | sort
Windows
Get-ChildItem app -Recurse -File
You should see seven __init__.py files, one inside each package directory.
What does each directory do?
- api/ holds your route handlers (the HTTP endpoints).
- models/ defines your database table structures.
- schemas/ defines the shape of data coming in and going out of your API.
- services/ contains the business logic that sits between routes and the database.
- database/ manages the connection to your database.
- core/ holds app-wide configuration settings.
- Open the aura-os-backend folder in VS Code by clicking File in the top menu bar.
- Click Open Folder.
- Navigate to your Desktop, select the aura-os-backend folder, then click Open.
The Explorer panel on the left now shows the app folder with all six subdirectories and their __init__.py files.
Don't see the folders in the Explorer?
Make sure you opened the aura-os-backend folder itself, not its parent (Desktop). The Explorer should show app at the top level. If you see aura-os-backend as a subfolder instead, go to File > Open Folder again and select the correct folder.
help me find my project folder in VS Code
Set up the virtual environment and install dependencies
A virtual environment gives your project its own isolated set of Python packages. This keeps your project's dependencies separate from any other Python projects on your machine.
- In the VS Code Explorer panel, click the New File icon at the top of the panel (the page icon with a plus sign).
- Name the file requirements.txt and press Enter.
- Paste the following pinned dependencies into the file:
fastapi==0.141.1
uvicorn==0.52.4
sqlalchemy==2.0.52
pydantic-settings==2.15.0
What are these packages?
- FastAPI is the web framework that handles your API routes, request validation, and auto-generated documentation.
- uvicorn is the server that runs your FastAPI app and serves it over HTTP.
- SQLAlchemy is the database toolkit that lets your Python code talk to databases through an ORM instead of raw SQL.
- pydantic-settings reads configuration from environment variables with type validation, following the 12-factor app pattern.
- Save requirements.txt by pressing Cmd+S (macOS) or Ctrl+S (Windows).
- Open a new terminal in VS Code by selecting Terminal > New Terminal from the top menu.
The terminal should open inside your aura-os-backend folder automatically, since that is the folder VS Code has open.
- Create a virtual environment by running this command:
macOS
python3 -m venv venv
Windows
python -m venv venv
- Activate the virtual environment by running:
macOS
source venv/bin/activate
Windows
venv\Scripts\activate
You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.
Windows: seeing a script execution policy error?
PowerShell may block scripts by default. Run the following command to allow scripts for the current session, then try the activate command again:
Set-ExecutionPolicy -Scope Process -ExecutionPolicy Bypass
help me fix my virtual environment activation
- Install all four dependencies by running:
pip install -r requirements.txt
You should see output showing each package being downloaded and installed, ending with a line that says Successfully installed followed by the package names.
Before you run the next command, do you think Python will be able to import all four packages you just installed?
- Verify the installation by running:
python -c "import fastapi; print('FastAPI', fastapi.__version__)"
You should see FastAPI 0.141.1 printed in the terminal. This confirms Python can find the package and it is the correct version.
Seeing ModuleNotFoundError?
Check that (venv) appears in your terminal prompt. If it does not, the virtual environment is not active and pip install may have installed packages to the wrong location. Activate the environment and run pip install -r requirements.txt again.
help me fix my pip installation
✔️ Awesome, I've got everything!
Double check you have saved requirements.txt and your virtual environment is active ((venv) in the prompt).
ⓧ I'd like to double check the full code
The only file with content at this point is requirements.txt. All seven __init__.py files should be empty. Here is the full requirements.txt:
fastapi==0.141.1
uvicorn==0.52.4
sqlalchemy==2.0.52
pydantic-settings==2.15.0
Your folder structure should look like this:
aura-os-backend/
requirements.txt
venv/
app/
__init__.py
api/
__init__.py
core/
__init__.py
database/
__init__.py
models/
__init__.py
schemas/
__init__.py
services/
__init__.py
Your Python environment is ready and your project scaffold is in place. Next up, you will write your first API endpoints and see them live in the browser.
Build Your First Endpoints
Your Python environment is ready and the project structure is in place. Now you need proof that everything actually works.
The fastest way to prove it is to get a running server you can interact with in the browser. FastAPI generates interactive API documentation automatically, so the moment your server starts, you get a visual playground to test every endpoint. You will also wire up quick device endpoints backed by a simple Python list.
In this step, get ready to:
- Build a health check endpoint and device registration endpoints in app/main.py.
- Start the server and test your endpoints through auto-generated Swagger UI docs.
- Discover what happens to in-memory data when the server restarts.
Create your FastAPI app and health endpoint
Every FastAPI project starts with an app instance. This is the object that receives every incoming request and routes it to the right function.
- In the VS Code Explorer sidebar, right-click on the app folder and select New File.
- Name the file main.py and press Enter.
- Copy the following code into main.py:
from fastapi import FastAPI
# Create the FastAPI app instance
app = FastAPI(title="AURA OS Backend", version="0.1.0")
# Temporary in-memory storage for devices
devices = []
# Health check endpoint to verify the server is running
@app.get("/health")
def health_check():
return {"status": "healthy", "service": "aura-os-backend"}
What does this code do?
FastAPI(title=..., version=...) creates the app object that handles all incoming HTTP requests. The title and version show up in the auto-generated documentation page.
devices = [] is a plain Python list that will temporarily hold device data in memory. You will use this shortly for the device endpoints.
@app.get("/health") registers a GET endpoint at the /health path. It returns a simple JSON object confirming the server is alive.
- Save main.py.
- Start the server by running this command in your VS Code terminal:
uvicorn app.main:app --reload
What does this command do?
uvicorn is the server that runs your FastAPI app. app.main:app tells it to look inside the app folder, open main.py, and find the variable called app.
The --reload flag watches your files for changes and automatically restarts the server whenever you save. This means you will not need to manually restart for every code edit.
You should see output in the terminal ending with a line like:
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
Server not starting?
- If you see command not found, your virtual environment may not be active. Run venv\Scripts\activate (Windows) or source venv/bin/activate (macOS/Linux) and try again.
- If you see Error loading ASGI app, check that main.py is inside the app folder, not in the project root.
Help me debug my uvicorn server startup error.
- Open your browser and go to http://localhost:8000/docs.
- Click on the GET /health endpoint to expand it.
- Click Try it out.
- Click Execute.
You should see a JSON response with {"status": "healthy", "service": "aura-os-backend"} in the response body.
What is Swagger UI?
The page at /docs is Swagger UI, an interactive documentation page that FastAPI generates automatically from your code. Every endpoint you create appears here with a built-in form to test it. No need for a separate HTTP client.
Add device registration endpoints
A health endpoint proves the server runs, but your API needs to do real work. The next two endpoints let you register devices and retrieve the full list.
- Switch back to app/main.py in VS Code.
- Add the following code below the health_check function:
# Register a new device by appending it to the list
@app.post("/devices")
def register_device(device: dict):
devices.append(device)
return device
# Return all registered devices
@app.get("/devices")
def list_devices():
return devices
What does this code do?
register_device accepts any JSON object the caller sends and appends it to the devices list. Using dict as the type hint means FastAPI accepts any shape of JSON body for now.
list_devices returns the entire devices list as a JSON array. Together, these two endpoints form a minimal CRUD cycle: create and read.
- Save main.py. The --reload flag automatically restarts the server with your changes.
New endpoints not showing up in Swagger?
- Check your terminal for error messages. A syntax error in main.py prevents the reload from completing.
- Make sure you pasted the new code below the health_check function, not inside it. Each function should start at the same indentation level (no extra indent).
- Refresh the /docs page in your browser. Sometimes the browser caches the old page.
Help me figure out why my new endpoints aren't appearing.
✔️ Awesome, I've got everything!
Great. Make sure you have saved main.py and that your terminal shows a successful reload with no errors.
ⓧ I'd like to double check the full code
Here is the complete app/main.py file at this point:
from fastapi import FastAPI
app = FastAPI(title="AURA OS Backend", version="0.1.0")
devices = []
@app.get("/health")
def health_check():
return {"status": "healthy", "service": "aura-os-backend"}
@app.post("/devices")
def register_device(device: dict):
devices.append(device)
return device
@app.get("/devices")
def list_devices():
return devices
Now test the full flow in your browser.
- Refresh the Swagger UI page at http://localhost:8000/docs.
- Click on POST /devices to expand it.
- Click Try it out.
- Replace the request body with this JSON:
{
"device_name": "ESP32-Sensor-01",
"device_type": "esp32-s3"
}
- Click Execute.
You should see a 200 response with your device JSON echoed back.
- Register a second device by clicking Try it out again and sending this body:
{
"device_name": "ESP32-Sensor-02",
"device_type": "esp32-s3"
}
- Click Execute.
- Now click on GET /devices, then Try it out, then Execute.
You should see both devices returned in a JSON array.
Restart the server and check your data
You have two devices registered and the API is responding perfectly. Before moving on, test one more thing.
Before you restart, what do you think GET /devices will return after the server stops and starts again?
- Go to your VS Code terminal where uvicorn is running.
- Press Ctrl+C to stop the server.
- Restart the server by running the same command:
uvicorn app.main:app --reload
- Switch back to your browser and refresh the Swagger UI at http://localhost:8000/docs.
- Click on GET /devices, then Try it out, then Execute.
The response is an empty list: []. Both devices are gone.
Why did the data disappear?
The devices list lives in your Python process's memory. When you stop the server, that process ends and everything in memory is wiped clean. Restarting creates a brand new process with a fresh, empty list.
This is exactly the problem that databases solve. A database stores data on disk, so it survives restarts, crashes, and deployments.
You now have a running API with auto-generated docs, and you have experienced the fundamental limitation of in-memory storage firsthand. Next up, you will replace that temporary list with a real database so your device data survives restarts.
Add Persistent Device Storage
You just watched your registered devices vanish the moment you restarted the server. An in-memory list dies with the process, and that is exactly the problem a database solves.
In this step, you will add a persistent database layer using SQLAlchemy, an ORM that lets your Python code talk to a database through objects instead of raw SQL. You will start with SQLite (zero setup, ships with Python), and the architecture supports swapping to PostgreSQL by changing a single environment variable.
In this step, get ready to:
- Set up application configuration and a database session factory.
- Define the Device ORM model and Pydantic validation schemas.
- Wire up router-based endpoints and confirm data survives a server restart.
Set up application configuration and database session
Before connecting to a database, you need two foundational pieces: a settings module that holds your database URL, and a session module that creates the database engine and manages connections.
- Create a new file called app/core/config.py and add the following code:
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
# Application-level settings with sensible defaults
app_name: str = "AURA OS Backend"
database_url: str = "sqlite:///./aura.db"
api_v1_prefix: str = "/api/v1"
# Read env vars prefixed with AURA_ (e.g. AURA_DATABASE_URL)
model_config = {"env_prefix": "AURA_"}
# Single instance shared across the entire app
settings = Settings()
What does this code do?
The Settings class uses pydantic-settings to load configuration from environment variables. Every field has a default, so the app works out of the box.
The env_prefix of AURA_ means you can override any setting with an environment variable like AURA_DATABASE_URL. To switch from SQLite to PostgreSQL later, you only change that one variable.
- Save config.py.
- Verify the settings load by running this command in your terminal:
python -c "from app.core.config import settings; print(settings.database_url)"
You should see sqlite:///./aura.db printed in the terminal.
Seeing a ModuleNotFoundError?
Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt. If not, run venv\Scripts\activate again.
Also confirm you are running the command from the aura-os-backend folder, not from inside the app subfolder.
Help me debug my ModuleNotFoundError for pydantic_settings
- Now create app/database/session.py with the database engine and session factory:
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker
from app.core.config import settings
# SQLite needs this arg to work with multi-threaded ASGI servers
connect_args = {}
if settings.database_url.startswith("sqlite"):
connect_args["check_same_thread"] = False
# Create the database engine using the URL from settings
engine = create_engine(settings.database_url, connect_args=connect_args)
# Factory that produces new database sessions
SessionLocal = sessionmaker(bind=engine)
# Base class that all ORM models inherit from
class Base(DeclarativeBase):
pass
# Dependency that gives each request its own session
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
What does this code do?
- create_engine opens a connection pool to your database. The check_same_thread argument is a SQLite-specific setting needed because FastAPI handles multiple requests concurrently.
- SessionLocal is a factory. Each time you call it, you get a fresh database session for one request.
- Base is the parent class every ORM model will inherit from. It connects your Python classes to database tables.
- get_db() is a dependency function. FastAPI calls it automatically for each request, handing the endpoint a session and closing it when the request finishes.
What is dependency injection?
Dependency injection means FastAPI automatically provides things your endpoint needs (like a database session) instead of the endpoint creating them itself. You declare what you need in the function signature, and FastAPI handles the rest.
- Save session.py.
- Confirm the engine initializes by running this command:
python -c "from app.database.session import engine; print(engine)"
You should see Engine(sqlite:///./aura.db) printed in the terminal. Your configuration and database layers are wired up.
Seeing an ImportError?
Double-check that session.py is inside the app/database/ folder and that app/database/__init__.py exists (it can be empty). The __init__.py file tells Python this folder is a package.
Help me debug my ImportError for app.database.session
Define the Device model and Pydantic schemas
With the database layer in place, you need three things: an ORM model that maps a Python class to a database table, Pydantic schemas that validate incoming data and shape outgoing responses, and a service module that contains the business logic.
- Create app/models/device.py with the Device ORM model:
import uuid
from datetime import datetime
from sqlalchemy import String, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.session import Base
class Device(Base):
__tablename__ = "devices"
# Auto-generated UUID stored as a 36-character string
id: Mapped[str] = mapped_column(
String(36), primary_key=True, default=lambda: str(uuid.uuid4())
)
device_name: Mapped[str] = mapped_column(String(100))
device_type: Mapped[str] = mapped_column(String(50), default="esp32-s3")
firmware_version: Mapped[str | None] = mapped_column(String(50), nullable=True)
ip_address: Mapped[str | None] = mapped_column(String(45), nullable=True)
status: Mapped[str] = mapped_column(String(20), default="registered")
registered_at: Mapped[datetime] = mapped_column(
DateTime, server_default=func.now()
)
last_seen_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
# Relationship to telemetry readings (used in the next step)
telemetry: Mapped[list["TelemetryReading"]] = relationship(
back_populates="device"
)
What does this model do?
- __tablename__ tells SQLAlchemy the name of the database table this class maps to.
- Each Mapped field becomes a column in the devices table. The type annotation (str, str | None) controls whether the column is required or nullable.
- server_default=func.now() tells the database to generate the timestamp, not Python. This keeps timestamps consistent regardless of server timezone.
- The telemetry relationship is a forward reference to a model you will create in the next step. SQLAlchemy resolves it later when both models exist.
- Save device.py.
- Open app/models/__init__.py and add this import so SQLAlchemy discovers the model:
from app.models.device import Device
- Save __init__.py.
- Confirm the model loads by running this command:
python -c "from app.models.device import Device; print(Device.__tablename__)"
You should see devices printed. Your ORM model is ready.
- Create app/schemas/device.py with the Pydantic validation schemas:
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
# Schema for incoming request bodies (what the client sends)
class DeviceCreate(BaseModel):
device_name: str = Field(..., min_length=1, max_length=100)
device_type: str = Field(default="esp32-s3", max_length=50)
firmware_version: str | None = None
ip_address: str | None = None
# Schema for outgoing responses (what the API returns)
class DeviceResponse(BaseModel):
# Tells Pydantic to read data from ORM model attributes
model_config = ConfigDict(from_attributes=True)
id: str
device_name: str
device_type: str
firmware_version: str | None
ip_address: str | None
status: str
registered_at: datetime
last_seen_at: datetime | None
Why separate schemas from models?
The ORM model (Device) defines how data is stored in the database. The Pydantic schemas (DeviceCreate and DeviceResponse) define what the API accepts and returns.
DeviceCreate only includes fields the client provides (like device_name). Fields like id and registered_at are auto-generated on the server side, so they only appear in DeviceResponse.
- Save device.py in the schemas folder.
- Verify the schemas load by running:
python -c "from app.schemas.device import DeviceCreate; print(DeviceCreate.model_fields.keys())"
You should see dict_keys(['device_name', 'device_type', 'firmware_version', 'ip_address']) printed.
Build the service layer and wire up endpoints
The service layer contains your business logic. It sits between the API endpoints and the database, keeping your route handlers clean and focused on HTTP concerns.
- Create app/services/device_service.py with the device CRUD functions:
from sqlalchemy.orm import Session
from app.models.device import Device
from app.schemas.device import DeviceCreate
def create_device(db: Session, device_data: DeviceCreate) -> Device:
# Build a Device ORM instance from validated input
device = Device(
device_name=device_data.device_name,
device_type=device_data.device_type,
firmware_version=device_data.firmware_version,
ip_address=device_data.ip_address,
)
db.add(device)
db.commit()
db.refresh(device)
return device
def get_devices(db: Session) -> list[Device]:
return db.query(Device).all()
def get_device(db: Session, device_id: str) -> Device | None:
return db.query(Device).filter(Device.id == device_id).first()
What does this code do?
- create_device() takes validated Pydantic data, creates a Device ORM object, saves it to the database, and returns the saved record with its auto-generated id.
- db.refresh(device) reloads the object from the database so fields like registered_at (generated by the database) are populated on the Python object.
- get_devices() returns all devices. get_device() finds a single device by its UUID, returning None if it does not exist.
- Save device_service.py.
- Now create app/api/health.py to extract the health endpoint into its own router:
from fastapi import APIRouter
router = APIRouter()
@router.get("/health")
def health_check():
# Simple liveness check for the backend
return {"status": "healthy", "service": "aura-os-backend"}
- Save health.py.
- Verify both new files import cleanly by running:
python -c "from app.services.device_service import create_device; from app.api.health import router; print('Service and health router loaded')"
You should see Service and health router loaded printed.
- Create app/api/devices.py with the database-backed device endpoints:
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.database.session import get_db
from app.schemas.device import DeviceCreate, DeviceResponse
from app.services import device_service
# All routes in this router are prefixed with /devices
router = APIRouter(prefix="/devices", tags=["Devices"])
@router.post(
"/", response_model=DeviceResponse, status_code=status.HTTP_201_CREATED
)
def register_device(
device_data: DeviceCreate, db: Session = Depends(get_db)
):
return device_service.create_device(db, device_data)
@router.get("/", response_model=list[DeviceResponse])
def list_devices(db: Session = Depends(get_db)):
return device_service.get_devices(db)
@router.get("/{device_id}", response_model=DeviceResponse)
def get_device(device_id: str, db: Session = Depends(get_db)):
device = device_service.get_device(db, device_id)
if not device:
raise HTTPException(status_code=404, detail="Device not found")
return device
What does this code do?
- APIRouter(prefix="/devices") groups all device routes under /devices. The tags parameter organizes them in the Swagger UI.
- Depends(get_db) injects a database session into each endpoint. FastAPI calls get_db() automatically and closes the session when the request completes.
- response_model tells FastAPI to serialize the return value using the Pydantic schema. This ensures the JSON response only includes the fields defined in DeviceResponse.
- Save devices.py.
- Now replace the entire contents of app/main.py with the new router-based architecture:
from fastapi import FastAPI
from app.core.config import settings
from app.database.session import Base, engine
from app.api import devices, health
from app.models import Device # noqa: F401
# Create all database tables defined by ORM models
Base.metadata.create_all(bind=engine)
app = FastAPI(title=settings.app_name, version="0.1.0")
# Register routers with the app
app.include_router(health.router)
app.include_router(devices.router, prefix=settings.api_v1_prefix)
What changed in main.py?
- The in-memory devices = [] list and the inline route functions are gone. All endpoints now live in dedicated router files.
- Base.metadata.create_all(bind=engine) creates the devices table in SQLite if it does not already exist.
- The Device import with # noqa: F401 looks unused, but it ensures SQLAlchemy registers the model before create_all() runs. Without it, no table gets created.
- include_router() mounts each router. The devices router gets the /api/v1 prefix, so the device endpoints are now at /api/v1/devices.
- Save main.py.
✔️ Awesome, I've got everything!
Great. Make sure all your files are saved. Time to test.
ⓧ I'd like to double check the full code
Here is each file you created or modified in this step. Compare yours against these.
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
app_name: str = "AURA OS Backend"
database_url: str = "sqlite:///./aura.db"
api_v1_prefix: str = "/api/v1"
model_config = {"env_prefix": "AURA_"}
settings = Settings()
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker
from app.core.config import settings
connect_args = {}
if settings.database_url.startswith("sqlite"):
connect_args["check_same_thread"] = False
engine = create_engine(settings.database_url, connect_args=connect_args)
SessionLocal = sessionmaker(bind=engine)
class Base(DeclarativeBase):
pass
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
import uuid
from datetime import datetime
from sqlalchemy import String, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.session import Base
class Device(Base):
__tablename__ = "devices"
id: Mapped[str] = mapped_column(
String(36), primary_key=True, default=lambda: str(uuid.uuid4())
)
device_name: Mapped[str] = mapped_column(String(100))
device_type: Mapped[str] = mapped_column(String(50), default="esp32-s3")
firmware_version: Mapped[str | None] = mapped_column(String(50), nullable=True)
ip_address: Mapped[str | None] = mapped_column(String(45), nullable=True)
status: Mapped[str] = mapped_column(String(20), default="registered")
registered_at: Mapped[datetime] = mapped_column(
DateTime, server_default=func.now()
)
last_seen_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
telemetry: Mapped[list["TelemetryReading"]] = relationship(
back_populates="device"
)
from app.models.device import Device
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
class DeviceCreate(BaseModel):
device_name: str = Field(..., min_length=1, max_length=100)
device_type: str = Field(default="esp32-s3", max_length=50)
firmware_version: str | None = None
ip_address: str | None = None
class DeviceResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: str
device_name: str
device_type: str
firmware_version: str | None
ip_address: str | None
status: str
registered_at: datetime
last_seen_at: datetime | None
from sqlalchemy.orm import Session
from app.models.device import Device
from app.schemas.device import DeviceCreate
def create_device(db: Session, device_data: DeviceCreate) -> Device:
device = Device(
device_name=device_data.device_name,
device_type=device_data.device_type,
firmware_version=device_data.firmware_version,
ip_address=device_data.ip_address,
)
db.add(device)
db.commit()
db.refresh(device)
return device
def get_devices(db: Session) -> list[Device]:
return db.query(Device).all()
def get_device(db: Session, device_id: str) -> Device | None:
return db.query(Device).filter(Device.id == device_id).first()
from fastapi import APIRouter
router = APIRouter()
@router.get("/health")
def health_check():
return {"status": "healthy", "service": "aura-os-backend"}
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.database.session import get_db
from app.schemas.device import DeviceCreate, DeviceResponse
from app.services import device_service
router = APIRouter(prefix="/devices", tags=["Devices"])
@router.post(
"/", response_model=DeviceResponse, status_code=status.HTTP_201_CREATED
)
def register_device(
device_data: DeviceCreate, db: Session = Depends(get_db)
):
return device_service.create_device(db, device_data)
@router.get("/", response_model=list[DeviceResponse])
def list_devices(db: Session = Depends(get_db)):
return device_service.get_devices(db)
@router.get("/{device_id}", response_model=DeviceResponse)
def get_device(device_id: str, db: Session = Depends(get_db)):
device = device_service.get_device(db, device_id)
if not device:
raise HTTPException(status_code=404, detail="Device not found")
return device
from fastapi import FastAPI
from app.core.config import settings
from app.database.session import Base, engine
from app.api import devices, health
from app.models import Device # noqa: F401
Base.metadata.create_all(bind=engine)
app = FastAPI(title=settings.app_name, version="0.1.0")
app.include_router(health.router)
app.include_router(devices.router, prefix=settings.api_v1_prefix)
Now for the real test. Before you start the server, think about this: if you register a device and then restart the server, what do you expect to happen when you query the device list?
- Start the server by running:
uvicorn app.main:app --reload
- Open http://localhost:8000/docs in your browser.
- Expand the POST /api/v1/devices/ endpoint and click Try it out.
- Replace the request body with this JSON and click Execute:
{
"device_name": "living-room-sensor",
"device_type": "esp32-s3",
"firmware_version": "1.0.0"
}
You should see a 201 response with the device data, including an auto-generated id and registered_at timestamp.
- Stop the server by pressing Ctrl+C in the terminal.
- Restart the server by running:
uvicorn app.main:app --reload
- Go back to http://localhost:8000/docs and try the GET /api/v1/devices/ endpoint.
Your living-room-sensor is still there. The data survived the restart because it is stored in the aura.db file that now exists in your project root.
- Confirm the database file exists by running this in a second terminal (keep the server running):
ls aura.db
You should see aura.db listed. That file is your SQLite database, holding everything your API persists.
Device list still empty after restart?
- Make sure you replaced the entire contents of app/main.py. If the old in-memory routes are still there, the server uses those instead of the database.
- Check that app/models/__init__.py contains from app.models.device import Device. Without this import, create_all() does not know about the Device model and skips creating the table.
- Verify that you are hitting /api/v1/devices/ (with the /api/v1 prefix), not the old /devices path.
Help me debug why my devices are not persisting after a server restart
Why SQLAlchemy instead of raw SQL?
SQLAlchemy is an ORM (Object-Relational Mapper). Instead of writing SQL queries by hand, you work with Python objects. Your code says db.add(device) instead of INSERT INTO devices ....
The real payoff is portability. Your code talks to a generic database layer. Switching from SQLite to PostgreSQL means changing one environment variable, not rewriting queries.
Your devices now survive restarts. The database layer is solid. Next up, you will build the telemetry pipeline so your devices can report CPU, memory, and Wi-Fi readings back to the backend.
Ingest Telemetry Data
Your backend now persists devices across restarts. But a registered device that never reports its status is invisible to the rest of the system. Real edge devices need to continuously send readings so you know they are alive and healthy.
In this step, you'll build the telemetry pipeline. This is the core data flow that the ESP32-S3 firmware will use to push CPU usage, memory usage, Wi-Fi signal strength, and uptime readings to your backend.
In this step, get ready to:
- Create the TelemetryReading database model and link it to Device.
- Build Pydantic schemas and a service layer for telemetry ingestion.
- Wire up POST and GET telemetry endpoints and test the full flow.
Create the telemetry data model
Each telemetry reading captures a snapshot of a device's health at a single point in time. The TelemetryReading model stores CPU usage, memory usage, Wi-Fi signal strength, and uptime, linked back to the device that sent it.
- Create a new file called telemetry.py inside the app/models/ folder, then paste the following code:
import uuid
from datetime import datetime
from sqlalchemy import String, Float, Integer, DateTime, ForeignKey, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.session import Base
class TelemetryReading(Base):
__tablename__ = "telemetry_readings"
# Unique ID for each reading
id: Mapped[str] = mapped_column(
String(36), primary_key=True, default=lambda: str(uuid.uuid4())
)
# Link each reading to the device that sent it
device_id: Mapped[str] = mapped_column(
String(36), ForeignKey("devices.id")
)
# Sensor readings from the device
cpu_usage: Mapped[float | None] = mapped_column(Float, nullable=True)
memory_usage: Mapped[float | None] = mapped_column(Float, nullable=True)
wifi_rssi: Mapped[int | None] = mapped_column(Integer, nullable=True)
uptime_seconds: Mapped[int | None] = mapped_column(Integer, nullable=True)
recorded_at: Mapped[datetime] = mapped_column(
DateTime, server_default=func.now()
)
# Relationship back to the parent device
device: Mapped["Device"] = relationship(back_populates="telemetry")
What does this code do?
- device_id is a foreign key that ties each reading to a row in the devices table.
- Each sensor field (cpu_usage, memory_usage, wifi_rssi, uptime_seconds) is nullable because a device might not report every metric in every reading.
- The device relationship connects back to the Device model's telemetry relationship, completing the two-way link between devices and their readings.
The Device model already has a telemetry relationship pointing to TelemetryReading. Now you need to register the new model so SQLAlchemy knows to create its table.
- In app/models/__init__.py, find this line:
from app.models.device import Device
- Add a new import line directly below it, so the file looks like this:
from app.models.device import Device
from app.models.telemetry import TelemetryReading
- Save both app/models/telemetry.py and app/models/__init__.py.
- Check your terminal. The server should auto-reload without errors.
Seeing an import error?
- Make sure the file is named telemetry.py (not telemetry_reading.py) and lives inside app/models/.
- Confirm the import line in __init__.py reads from app.models.telemetry import TelemetryReading with a capital T and R.
Help me debug a model import error in my FastAPI project.
Build the telemetry schemas and service
With the database model in place, you need two more layers before the endpoints: Pydantic schemas to validate incoming data, and a service layer to handle the business logic.
- Create a new file called telemetry.py inside the app/schemas/ folder, then paste the following code:
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
class TelemetryCreate(BaseModel):
# Field constraints ensure values stay within realistic ranges
cpu_usage: float | None = Field(None, ge=0, le=100)
memory_usage: float | None = Field(None, ge=0, le=100)
wifi_rssi: int | None = Field(None, ge=-100, le=0)
uptime_seconds: int | None = Field(None, ge=0)
class TelemetryResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: str
device_id: str
cpu_usage: float | None
memory_usage: float | None
wifi_rssi: int | None
uptime_seconds: int | None
recorded_at: datetime
What do these constraints do?
The Field constraints reject invalid data before it reaches the database. CPU and memory usage must be between 0 and 100 (percentage). Wi-Fi RSSI must be between -100 and 0 (signal strength in dBm, where 0 is perfect and -100 is no signal). Uptime must be zero or positive.
If a device sends cpu_usage: 150, FastAPI automatically returns a 422 Unprocessable Entity error before your code ever runs.
- Save app/schemas/telemetry.py.
Now build the service that saves telemetry readings and keeps the device's heartbeat updated.
- Create a new file called telemetry_service.py inside the app/services/ folder, then paste the following code:
from datetime import datetime, timezone
from sqlalchemy.orm import Session
from app.models.device import Device
from app.models.telemetry import TelemetryReading
from app.schemas.telemetry import TelemetryCreate
def create_telemetry(
db: Session, device_id: str, data: TelemetryCreate
) -> TelemetryReading:
# Build a new reading from the incoming data
reading = TelemetryReading(
device_id=device_id,
cpu_usage=data.cpu_usage,
memory_usage=data.memory_usage,
wifi_rssi=data.wifi_rssi,
uptime_seconds=data.uptime_seconds,
)
db.add(reading)
# Update the device's last_seen_at and mark it online
device = db.query(Device).filter(Device.id == device_id).first()
if device:
device.last_seen_at = datetime.now(timezone.utc)
device.status = "online"
db.commit()
db.refresh(reading)
return reading
Why does saving a reading also update the device?
Every telemetry POST acts as a heartbeat. The create_telemetry function does two things in one transaction: it saves the sensor reading AND stamps the device's last_seen_at with the current UTC time.
This timestamp is how future components (like the Angular dashboard) determine whether a device is online or offline. If a device stops sending readings, its last_seen_at falls behind and the system knows it has gone dark.
- Save app/services/telemetry_service.py.
- Check the terminal. The server should reload without errors.
Server crashing on reload?
- Make sure the file is named telemetry_service.py (with an underscore) inside app/services/.
- Verify that the import from app.schemas.telemetry import TelemetryCreate matches the schema file you just created.
Help me debug a service import error in my FastAPI project.
Wire up the endpoints and test the full flow
The model, schemas, and service are ready. Now you need API endpoints that expose them and a registration in main.py to make them reachable.
- Create a new file called telemetry.py inside the app/api/ folder, then paste the following code:
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from app.database.session import get_db
from app.schemas.telemetry import TelemetryCreate, TelemetryResponse
from app.services import device_service, telemetry_service
# Nest telemetry routes under /devices/{device_id}/telemetry
router = APIRouter(
prefix="/devices/{device_id}/telemetry", tags=["Telemetry"]
)
@router.post(
"/", response_model=TelemetryResponse, status_code=201
)
def ingest_telemetry(
device_id: str,
data: TelemetryCreate,
db: Session = Depends(get_db),
):
# Verify the device exists before accepting telemetry
device = device_service.get_device(db, device_id)
if not device:
raise HTTPException(status_code=404, detail="Device not found")
return telemetry_service.create_telemetry(db, device_id, data)
What does this code do?
- The router prefix nests all telemetry routes under /devices/{device_id}/telemetry. This is a RESTful pattern where child resources live under their parent.
- The POST endpoint first checks that the device exists. If someone sends telemetry for a nonexistent device ID, they get a clear 404 instead of a cryptic database error.
- Save app/api/telemetry.py.
The router exists, but FastAPI does not know about it yet. You need to register it in main.py.
- Open app/main.py and replace its contents with this updated version:
from fastapi import FastAPI
from app.core.config import settings
from app.database.session import Base, engine
# Import the new telemetry router and model
from app.api import devices, health, telemetry
from app.models import Device, TelemetryReading # noqa: F401
Base.metadata.create_all(bind=engine)
app = FastAPI(title=settings.app_name, version="0.1.0")
app.include_router(health.router)
app.include_router(devices.router, prefix=settings.api_v1_prefix)
# Register the telemetry router under the same API prefix
app.include_router(telemetry.router, prefix=settings.api_v1_prefix)
Why import TelemetryReading here?
Importing TelemetryReading ensures SQLAlchemy registers the model with Base.metadata before create_all runs. Without this import, the telemetry_readings table would never be created.
- Save app/main.py.
- Switch to your browser and refresh http://localhost:8000/docs.
- You should see a new Telemetry section in the Swagger UI with a POST endpoint.
Time to test the POST endpoint. You will register a fresh device, then send it a telemetry reading.
- In the Swagger UI, expand the Devices section and click the POST /api/v1/devices/ endpoint.
- Click Try it out.
- Replace the request body with this JSON and click Execute:
{
"device_name": "living-room-sensor",
"device_type": "esp32-s3",
"firmware_version": "1.0.0"
}
- You should see a 201 response. Copy the id value from the response. You will need it to send telemetry.
- Now expand the Telemetry section and click the POST /api/v1/devices/{device_id}/telemetry/ endpoint.
- Click Try it out.
- Paste your device ID into the device_id field.
- Replace the request body with this sample telemetry and click Execute:
{
"cpu_usage": 42.5,
"memory_usage": 67.3,
"wifi_rssi": -55,
"uptime_seconds": 3600
}
- You should see a 201 response with your telemetry reading, including a generated id and recorded_at timestamp.
Getting a 404 Device not found?
- Make sure you copied the full device id from the POST response (a UUID like a1b2c3d4-...), not the device name.
- If you previously deleted aura.db, you need to register a new device first before sending telemetry.
Help me debug a 404 error when posting telemetry.
The POST endpoint works. Now add the GET endpoint so you can query a device's telemetry history.
- In app/services/telemetry_service.py, add this function below the create_telemetry function:
def get_device_telemetry(
db: Session, device_id: str, limit: int = 50
) -> list[TelemetryReading]:
# Return readings newest-first, with a configurable limit
return (
db.query(TelemetryReading)
.filter(TelemetryReading.device_id == device_id)
.order_by(TelemetryReading.recorded_at.desc())
.limit(limit)
.all()
)
What does this function do?
This query returns the most recent readings first (desc() on recorded_at) and caps the result set with a configurable limit (defaulting to 50). The dashboard will call this to display a device's recent health trend.
- Save app/services/telemetry_service.py.
- In app/api/telemetry.py, add this GET endpoint below the ingest_telemetry function:
@router.get("/", response_model=list[TelemetryResponse])
def get_telemetry(
device_id: str,
limit: int = Query(default=50, le=200),
db: Session = Depends(get_db),
):
# Verify device exists before querying
device = device_service.get_device(db, device_id)
if not device:
raise HTTPException(status_code=404, detail="Device not found")
return telemetry_service.get_device_telemetry(db, device_id, limit)
What does the Query parameter do?
The limit parameter uses Query(default=50, le=200) to cap how many readings a single request can fetch. Without this, a client could request millions of rows and overwhelm the server.
- Save app/api/telemetry.py.
✔️ Awesome, I've got everything!
Make sure you have saved all your files.
ⓧ I'd like to double check the full code
Scroll up and confirm each piece is present in your files, in order:
- app/api/telemetry.py: imports and router setup, then the ingest_telemetry POST function, then the get_telemetry GET function.
- app/services/telemetry_service.py: imports, then the create_telemetry function, then the get_device_telemetry function.
- app/models/telemetry.py: the full TelemetryReading class with all six columns and the device relationship.
- app/schemas/telemetry.py: the TelemetryCreate and TelemetryResponse classes.
- app/main.py: imports include telemetry in the API import line and TelemetryReading in the models import line, plus the include_router call for the telemetry router.
- app/models/__init__.py: imports both Device and TelemetryReading.
Before you test the GET endpoint, think about what you expect to see. You POSTed one telemetry reading earlier. What should the GET response contain?
- Refresh http://localhost:8000/docs in your browser.
- Expand the Telemetry section. You should now see both the POST and GET endpoints.
- Click the GET /api/v1/devices/{device_id}/telemetry/ endpoint.
- Click Try it out, paste the same device ID, and click Execute.
You should see a JSON array containing the telemetry reading you posted earlier, with your CPU, memory, Wi-Fi, and uptime values.
One more check. The create_telemetry function should have updated the device's last_seen_at timestamp.
- Expand the Devices section and click GET /api/v1/devices/{device_id}.
- Click Try it out, paste the same device ID, and click Execute.
- In the response, last_seen_at should now show a timestamp (instead of null) and status should read "online".
Still seeing null for last_seen_at?
- Make sure you are querying the same device ID you sent telemetry to, not a different device.
- Check that your create_telemetry function includes the lines that update device.last_seen_at and device.status before the db.commit() call.
Help me debug why last_seen_at is not updating.
Secret mission
Device Health Status
Your backend can register devices and store telemetry, but it has no way to answer the simplest operational question: is this device online right now? In this secret mission, you will add a health endpoint that calculates online or offline status from the most recent telemetry timestamp and returns a full health summary.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing costs or cloud resources to worry about.
Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- The aura-os-backend project folder on your Desktop (includes all source code, the venv virtual environment, and the aura.db SQLite database).
- Python 3.13.15 installed on your system.
- The uvicorn development server (currently running in your terminal).
Keep everything running
No action needed. Choose this if you're still actively building on the AURA OS backend or plan to add more features soon.
- Your uvicorn server keeps running and serving requests at http://localhost:8000/docs.
- Your aura.db database retains all registered devices and telemetry data.
- Your virtual environment and installed packages stay ready in the venv folder.
Pause - I'll come back to this later
Shut down the running server to free up your terminal. Your project files and database stay on disk, ready for you to pick back up.
- Switch back to your VS Code terminal where uvicorn is running.
- Press Ctrl+C to stop the server.
- Deactivate the virtual environment by running:
deactivate
When you come back, reactivate the environment and restart the server:
- Open a terminal in your aura-os-backend folder.
- Activate the virtual environment with venv\Scripts\activate (Windows) or source venv/bin/activate (macOS/Linux).
- Start the server with uvicorn app.main:app --reload.
Delete - I don't want to use this again
Remove all project files, the database, and the virtual environment. This frees up disk space and leaves no project traces.
- Switch back to your VS Code terminal where uvicorn is running.
- Press Ctrl+C to stop the server.
- Deactivate the virtual environment by running:
deactivate
- Navigate to your Desktop and delete the project folder.
macOS
- Run these commands to delete the project folder:
cd ~/Desktop
rm -rf aura-os-backend
Windows
- Run these commands to delete the project folder:
cd ~\Desktop
rmdir /s /q aura-os-backend
- Confirm the folder is gone by running:
ls
You should no longer see aura-os-backend in the listing.
What about Python?
Python 3.13.15 stays installed on your system. If you want to remove it, open Settings on Windows, go to Apps, find Python 3.13.15 in the list, and click Uninstall.
If you plan to do more Python projects, keep it installed.
Nice Work!
Nice Work!
Nice work! You just built a production-ready backend API for the AURA OS device management system.
Your backend is the foundation every future AURA OS project plugs into. The ESP32-S3 firmware will POST to these endpoints, the Angular dashboard will GET from them, and the OTA system will manage firmware through them.
You've learned how to:
- Build a modular FastAPI backend with a clean layered architecture (api, models, schemas, services, database, core) that mirrors real-world Python project structure.
- Set up SQLAlchemy with a database abstraction layer that uses SQLite now but can switch to PostgreSQL by changing one environment variable. You experienced firsthand why persistent storage matters by watching in-memory data vanish on restart.
- Create device registration and telemetry ingestion endpoints with Pydantic validation, and browse every endpoint live through auto-generated Swagger UI at /docs.
- Secret Mission: An optional challenge to push your skills further by adding a device health status endpoint.
Ready to quiz yourself?