The service layer (src/service/) is the business-logic tier of the gateway. It sits between the HTTP/Socket.IO routes and the SQLAlchemy models plus every external system the gateway talks to: drones over rosbridge, Janus video rooms, AWS (S3/EC2/ECR), Docker, Stripe, and SMTP. Routes stay thin — parse the request, check auth, delegate to a service, serialize the result. There is no Flask dependency-injection container here. Every service is instantiated once at import time as a module-global in the composition root src/application/app.py, and route blueprints import those singletons by name. This page maps that wiring, the two lazy-initialization escape hatches (get_service and get_vpn_service), and the DroneService → Physical/SITL inheritance the factory returns.
Deep dives on each service live on sibling pages — see DroneControlService & rosbridge dispatch, SITL lifecycle, video rooms, assets & archives, executions & reports, billing, and platform services. This page is only about how they are constructed and injected.

Composition root: app.py

Importing src.application.app builds the entire singleton graph as a side effect. Order matters because later services take earlier ones as constructor arguments (drone_control_service and asset_service feed into physical_drone_service, for example).
src/application/app.py
video_service = VideoService()
email_service = EmailService()
user_service = UserService(db, Bcrypt(Flask(__name__)), email_service)
generic_drone_service = DroneService(db)
generic_mission_service = MissionService(db)
point_service = PointService(db)
geofence_service = GeofenceService(db)

drone_control_service = DroneControlService(generic_drone_service, generic_mission_service)

asset_service = AssetService(db)
physical_drone_service = PhysicalDroneService(db, video_service, drone_control_service, asset_service)
SingletonClassConstructed withNotes
video_serviceVideoServiceJanus VideoRoom management
email_serviceEmailServiceSMTP wrapper, shared by user + report services
user_serviceUserServicedb, Bcrypt(...), email_serviceAlso aliased as generic_user_service
generic_drone_serviceDroneServicedbBase CRUD; used directly by routes for lookups
generic_mission_serviceMissionServicedbAliased in routes as mission_service
point_servicePointServicedb
geofence_serviceGeofenceServicedb
drone_control_serviceDroneControlServicegeneric_drone_service, generic_mission_serviceBuilt without SocketIO (attached later — see below)
asset_serviceAssetServicedb
physical_drone_servicePhysicalDroneServicedb, video_service, drone_control_service, asset_serviceEager; the physical entry in the factory registry
subscription_serviceSubscriptionServicedbStripe billing
calendar_serviceCalendarServicedb
execution_serviceExecutionServicedb
report_serviceReportServicedb, email_service
archive_serviceArchiveServicedb
log_analysis_serviceLogAnalysisServicepymavlink .bin parsing
Not every service is constructed here. IsaacSimService is instantiated directly in src/routes/isaac_sim_routes.py:16, not in app.py. IPService is used statically (no instance). VPN_Service is built lazily via get_vpn_service(). And AWSInstanceService / InstanceSchedulerService are dormant — defined but never instantiated anywhere in the current codebase.

Dependency injection into routes

“DI” here is just Python module imports. Blueprints pull the singletons they need straight off app.py:
src/routes/drone_routes.py
from src.application.app import (
    drone_control_service,
    generic_drone_service,
    get_service,
    get_vpn_service,
    physical_drone_service,
    subscription_service,
)
Every blueprint follows the same shape (user_routes imports user_service, billing_routes imports subscription_service, execution_routes imports the execution/report/asset trio, and so on). To swap an implementation for a test, you patch the module global — there is no registry to override.

Startup wiring order (load-bearing)

DroneControlService is constructed in app.py without a SocketIO instance, because the gevent SocketIO object does not exist yet. src/main.py creates SocketIO, wires the socket routes, and only then attaches it back onto the already-built service:
src/main.py
from src.application import app as app_module

socket_routes.init_socket_routes(socketio, app_module.drone_control_service)   # ~line 240
...
app_module.drone_control_service.socketio = socketio                           # line 259
This ordering is required: DroneControlService only registers the telemetry-emit callback on a connection when self.socketio is truthy, so telemetry cannot flow until line 259 runs. See Connection Pool & Startup Wiring for the full sequence.

The get_service(drone_type) factory

Drone lifecycle operations (create, update, delete) are polymorphic across drone types, so routes never reference a concrete drone service directly. Instead they call get_service(drone_type) and get back the right DroneService subclass.
src/application/app.py
drone_services = {DroneTypes.physical.value: physical_drone_service}

def get_service(drone_type):
    if drone_type == DroneTypes.sitl.value and DroneTypes.sitl.value not in drone_services:
        if settings.ENABLE_SITL:
            drone_services[DroneTypes.sitl.value] = SITLDroneService(
                db, video_service, drone_control_service, asset_service
            )
        else:
            logger.warning("SITL service requested but SITL is not enabled.")
    drone_service = drone_services.get(drone_type)
    if drone_service:
        return drone_service
    raise UnknownDroneTypeException(drone_type)
The only valid keys are the DroneTypes enum values "physical" and "sitl" (src/utils/drone_types.py). Anything else raises UnknownDroneTypeException (src/utils/unknown_drone_type.py).
1

Physical is eager

physical_drone_service is created at import time and pre-seeded into the drone_services registry, so get_service("physical") is a plain dict lookup.
2

SITL is lazy behind ENABLE_SITL

SITLDroneService is only constructed on the first get_service("sitl") call, and only when settings.ENABLE_SITL is true (default true, from ENABLE_SITL env var). Its constructor connects to Docker (local docker.from_env() or a remote host over the jumphost tunnel), so this lazy pattern avoids paying that cost when SITL is disabled. Once built, it is cached in drone_services for the rest of the process.
3

Disabled SITL falls through to an error

If ENABLE_SITL is false, the factory logs a warning, never caches an instance, and the final drone_services.get("sitl") returns NoneUnknownDroneTypeException. Callers surface this as a failed drone creation.
Whether SITL operations actually run is a second, separate gate. Even when ENABLE_SITL builds the service, SITLDroneService.__init__ sets is_server_environment = True (and disables Docker orchestration) when the process is not local and not using remote Docker — i.e. not IS_LOCAL_ENVIRONMENT and not REMOTE_DOCKER_ENABLED. So ENABLE_SITL=true on a plain server env yields a service that exists but refuses to spawn containers. See SITL Drone Lifecycle.

get_vpn_service() lazy init

VPN_Service is the one supporting service deliberately kept out of the eager block. It builds a boto3 S3 client in its constructor, and that client binds to settings.REGION — which must be loaded and validated first. The lazy singleton guarantees the region is set before the client is created:
src/application/app.py
_vpn_service_instance = None

def get_vpn_service():
    global _vpn_service_instance
    if _vpn_service_instance is None:
        _vpn_service_instance = VPN_Service()
    return _vpn_service_instance
Routes call get_vpn_service() at request time (drone_routes.py, vpn_routes.py) rather than importing an instance. It presigns per-user / per-drone WireGuard configs and proxies VPN status checks — details on VPN IP Authentication & Jumphost Routing.

DroneService → Physical / SITL inheritance

The factory always returns a DroneService subclass. The base class owns generic CRUD and the rich lookup surface; the two subclasses override only the lifecycle methods that differ.
ConcernDroneService (base)PhysicalDroneServiceSITLDroneService
save()Sync; 3× retry in a nested tx; generates a 10-digit activation_token only when type == "physical"async; allocates a real IP via IPService.get_drone_ip() (ip == mac), calls super().save, then creates a Janus room + JWT room tokenasync; enforces USER_SITL_MAX_COUNT, spins up the 3-container Docker stack, saves with ip == mac == container_name and port = 9090 + n
delete()Sync DB delete by (id, user_id)async; deletes S3 assets → Janus room → DB rowasync; deletes S3 assets → stops core/gamepad/SITL containers (reverse order) → DB row
activate()Clears activation_token(inherits base)
Lookups (get_drone_by_id, _by_ip, _by_token, counts)✅ owns allinheritedinherited
Constructor argsdbdb, video_service, drone_control_service, asset_servicesame four args
Because save()/delete() are async on the subclasses but sync on the base, routes await the factory result (await drone_service.save(data) in drone_routes.py:111). The route also branches on return shape: SITL save() returns a dict wrapping the drone, physical save() returns the drone dict directly.
update() lives only on the base class and is intentionally not IP-aware — there is an explicit # NOTE: no IP updates comment. Routes call get_service(drone_type).update(data) regardless of type; both subclasses inherit it unchanged.

Adding a new drone type or service

1

Add the enum value

Extend DroneTypes in src/utils/drone_types.py. The factory keys strictly off these enum values, so an unlisted string can never resolve.
2

Implement the subclass

Subclass DroneService and override save/delete (mark them async to match how routes await them). Reuse the base lookups and generate_activation_token().
3

Register it in the factory

Either eager-seed it into the drone_services dict next to physical (like PhysicalDroneService), or add a lazy branch in get_service behind its own feature flag (like SITL). Prefer lazy if the constructor touches an external system (Docker, boto3, a socket).
4

Wire supporting singletons in app.py

If it is a plain supporting service (not a drone type), instantiate it in the app.py composition block in dependency order, then import the singleton in the consuming blueprint. If its constructor needs a late-bound resource (region, SocketIO), follow the get_vpn_service lazy pattern or the main.py post-init attachment pattern instead.
For the environment flags referenced above (ENABLE_SITL, DEPLOYMENT_ENVIRONMENT, REMOTE_DOCKER_ENABLED, REGION) and how startup validation fails fast, see Gateway Environment Variables and Startup, Validation & Composition Root.