Справочник API¶

Всеобъемлющая документация API для бэкенд сервисов CulicidaeLab Server.

Обзор¶

CulicidaeLab Server предоставляет RESTful API, построенный с FastAPI для исследования комаров, наблюдения и анализа данных. API предлагает endpoints для идентификации видов, отслеживания болезней, управления географическими данными и записи наблюдений.

Базовый URL¶

http://localhost:8000/api/v1

Аутентификация¶

В настоящее время API не требует аутентификации для большинства endpoints. Будущие версии могут реализовать аутентификацию по API ключу или OAuth2.

Формат ответа¶

Все ответы API следуют согласованному JSON формату с соответствующими HTTP кодами состояния:

200 OK: Успешный запрос
404 Not Found: Ресурс не найден
422 Unprocessable Entity: Ошибка валидации
500 Internal Server Error: Ошибка сервера

Интерактивная документация API¶

Для интерактивного исследования API посетите автоматически сгенерированную документацию:

Swagger UI: http://localhost:8000/docs
ReDoc: http://localhost:8000/redoc

Модули API¶

Основное приложение¶

`backend.main` ¶

Main application module for CulicidaeLab API backend.

This module initializes and configures the FastAPI application, including: - Application startup and shutdown lifecycle management - CORS middleware configuration - Static file serving setup - API router registration - Health check and root endpoints

The application serves as the main entry point for the CulicidaeLab API server, providing endpoints for species identification, disease tracking, geographic data, and observation management.

Example

python -m backend.main

Starts the server on http://localhost:8000 ¶

Or run with custom host/port¶

uvicorn backend.main:app --host 0.0.0.0 --port 8000 --reload

`api_router_prefix = ''` `module-attribute` ¶

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

`backend_dir = 'C:\\Users\\lenova\\culicidaelab-server\\backend'` `module-attribute` ¶

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

`settings = AppSettings(APP_NAME='CulicidaeLab API', API_V1_STR='/api', DATABASE_PATH='backend/data/.lancedb', SAVE_PREDICTED_IMAGES='1', BACKEND_CORS_ORIGINS=['http://localhost:8765', 'http://127.0.0.1:8765'])` `module-attribute` ¶

Application settings configuration for CulicidaeLab API backend.

This class defines all configuration parameters for the CulicidaeLab API server, using Pydantic BaseSettings for environment variable support and validation. Settings can be overridden via environment variables with the CULICIDAELAB_ prefix.

Attributes:

Name	Type	Description
`APP_NAME`	`str`	Name of the application displayed in API documentation.
`API_V1_STR`	`str`	Base path prefix for API version 1 endpoints.
`DATABASE_PATH`	`str`	File system path to the LanceDB database directory.
`SAVE_PREDICTED_IMAGES`	`str \| bool`	Whether to save predicted images to disk.
`BACKEND_CORS_ORIGINS`	`list[str]`	List of allowed CORS origins for frontend access.

Example

settings = AppSettings() print(f"App name: {settings.APP_NAME}") App name: CulicidaeLab API print(f"Database path: {settings.DATABASE_PATH}") Database path: .lancedb

`static_dir = 'C:\\Users\\lenova\\culicidaelab-server\\backend\\static'` `module-attribute` ¶

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

`lifespan(app: FastAPI)` ¶

Manages application startup and shutdown lifecycle.

This context manager handles the initialization of database connections, cache loading, and other startup tasks when the FastAPI application starts, and cleanup when it shuts down.

Parameters:

Name	Type	Description	Default
`app`	`FastAPI`	The FastAPI application instance.	required

Yields:

Name	Type	Description
`None`		Control is yielded back to FastAPI after startup initialization.

Example

@asynccontextmanager async def lifespan(app: FastAPI): # Initialize resources db = await init_database() app.state.db = db yield # Cleanup resources await db.close()

`read_root()` `async` ¶

`health_check()` `async` ¶

`get_db()` ¶

Establish a connection to the LanceDB database.

This function creates a cached connection to the LanceDB database using the path specified in the application settings. The connection is cached to avoid repeated connection overhead.

Returns:

Type	Description
	lancedb.DBConnection: A connection object to the LanceDB database.

Raises:

Type	Description
`Exception`	If the database connection fails, the original exception is re-raised after logging the error.

Example

db_connection = get_db() print(f"Connected to: {db_connection}")

`load_all_datasource_translations(db: object, supported_langs: list[str]) -> dict[str, dict[str, str]]` ¶

Load all data source translations from the database for multiple languages.

This function queries the data_sources table and extracts translations for all supported languages, using English as a fallback for missing translations. Returns an empty dictionary if the query fails.

Parameters:

Name	Type	Description	Default
`db`	`object`	The database connection object.	required
`supported_langs`	`list[str]`	A list of language codes to load (e.g., ['en', 'ru', 'es']).	required

Returns:

Type	Description
`dict[str, dict[str, str]]`	dict[str, dict[str, str]]: A nested dictionary structured as {language_code: {source_id: translated_name}}. Returns an empty dictionary if the query fails. Missing translations fall back to English or the source ID itself.

Example

db = get_db() translations = load_all_datasource_translations(db, ["en", "ru"]) print(translations["en"]["gbif"]) # "GBIF" print(translations["ru"]["gbif"]) # "GBIF"

`load_all_region_translations(db: object, supported_langs: list[str]) -> dict[str, dict[str, str]]` ¶

Load all region translations from the database for multiple languages.

This function queries the regions table and extracts translations for all supported languages, using English as a fallback for missing translations. The data is loaded once at application startup for performance.

Parameters:

Name	Type	Description	Default
`db`	`object`	The database connection object.	required
`supported_langs`	`list[str]`	A list of language codes to load (e.g., ['en', 'ru', 'es']).	required

Returns:

Type	Description
`dict[str, dict[str, str]]`	dict[str, dict[str, str]]: A nested dictionary structured as {language_code: {region_id: translated_name}}. Missing translations fall back to English or the region ID itself.

Example

db = get_db() translations = load_all_region_translations(db, ["en", "ru"]) print(translations["en"]["us"]) # "United States" print(translations["ru"]["us"]) # "Соединенные Штаты"

`load_all_species_names(db: object) -> list[str]` ¶

Load a sorted list of all unique species scientific names from the database.

This function queries the species table and extracts all unique scientific names, returning them as a sorted list. Returns an empty list if the query fails.

Parameters:

Name	Type	Description	Default
`db`	`object`	The database connection object.	required

Returns:

Type	Description
`list[str]`	list[str]: A sorted list of unique species scientific names. Returns an empty list if the query fails or no species are found.

Example

db = get_db() species_names = load_all_species_names(db) print(len(species_names)) # Number of species print(species_names[:3]) # First 3 species names

Управление видами¶

API видов предоставляет endpoints для получения информации о видах комаров, включая детальные данные о видах и информацию о переносчиках болезней.

`backend.routers.species` ¶

Species router module for the Culicidae Lab Server.

This module provides FastAPI route handlers for mosquito species-related operations in the Culicidae Lab application. It handles retrieval of species information, including species lists, detailed species information, and disease vector species.

The router integrates with the species service layer to perform database queries and return properly formatted responses according to the defined Pydantic schemas.

Routes

GET /species: Retrieve a paginated list of mosquito species with optional search
GET /species/{species_id}: Retrieve detailed information for a specific species
GET /vector-species: Retrieve species that are known disease vectors

Example

The router is typically mounted in the main FastAPI application:

from fastapi import FastAPI
from backend.routers.species import router as species_router

app = FastAPI()
app.include_router(species_router, prefix="/api/v1")

`SpeciesBase` ¶

Base model for species information.

Contains core species identification fields used across different species-related models.

`model_config = {}` `class-attribute` ¶

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

`SpeciesDetail` ¶

Detailed species model with extended information.

Extends the base species model with additional descriptive fields for comprehensive species information.

`model_config = {}` `class-attribute` ¶

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

`SpeciesListResponse` ¶

Response model for paginated species lists.

Contains the total count and list of species for API responses.

`model_config = {}` `class-attribute` ¶

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

`get_species_list_endpoint(request: Request, db: DBConnection = Depends(get_db), search: str | None = Query(None), limit: int = Query(50), lang: str = Query(en))` `async` ¶

Retrieve a list of mosquito species, optionally filtered by a search term.

This endpoint allows clients to fetch a paginated list of mosquito species from the database. The search functionality supports partial matching against scientific names and common names in English and Russian.

Parameters:

Name	Type	Description	Default
`request`	`Request`	The FastAPI request object used for URL construction.	required
`db`	`DBConnection`	LanceDB database connection for querying species data.	`Depends(get_db)`
`search`	`str \| None`	Optional search term to filter species by name. Supports partial matching against scientific names and common names in multiple languages.	`Query(None)`
`limit`	`int`	Maximum number of species to return (1-200). Defaults to 50.	`Query(50)`
`lang`	`str`	Language code for response localization (e.g., 'en', 'es', 'ru').	`Query(en)`

Returns:

Name	Type	Description
`SpeciesListResponse`		A response containing the count of species and the list of species matching the criteria.

Example

Get the first 25 mosquito species in English:

import httpx

async with httpx.AsyncClient() as client:
    response = await client.get(
        "http://localhost:8000/api/v1/species",
        params={"limit": 25, "lang": "en"}
    )
    species_data = response.json()
    print(f"Found {species_data['count']} species")

Search for species containing "aedes":

response = await client.get(
    "http://localhost:8000/api/v1/species",
    params={"search": "aedes", "limit": 10, "lang": "en"}
)

`get_species_detail_endpoint(species_id: str, request: Request, region_cache: dict[str, dict[str, str]] = Depends(get_region_cache), db: DBConnection = Depends(get_db), lang: str = Query(en))` `async` ¶

Retrieve detailed information for a specific mosquito species by ID.

This endpoint fetches comprehensive information about a single mosquito species including its scientific name, common names in multiple languages, vector status, habitat preferences, geographic regions, and associated diseases. The response includes translated content based on the requested language.

Parameters:

Name	Type	Description	Default
`species_id`	`str`	The unique identifier for the species (e.g., 'aedes-aegypti').	required
`request`	`Request`	The FastAPI request object used for URL construction.	required
`region_cache`	`dict[str, dict[str, str]]`	Pre-loaded cache of region translations for localization.	`Depends(get_region_cache)`
`db`	`DBConnection`	LanceDB database connection for querying species data.	`Depends(get_db)`
`lang`	`str`	Language code for response localization (e.g., 'en', 'es', 'ru').	`Query(en)`

Returns:

Name	Type	Description
`SpeciesDetail`		Detailed information about the requested species including scientific name, common names, descriptions, characteristics, habitat preferences, geographic regions, and related diseases.

Raises:

Type	Description
`HTTPException`	If the species with the given ID is not found, returns a 404 status code with detail message "Species not found".

Example

Get detailed information for Aedes aegypti in English:

import httpx

async with httpx.AsyncClient() as client:
    response = await client.get(
        "http://localhost:8000/api/v1/species/aedes-aegypti",
        params={"lang": "en"}
    )
    if response.status_code == 200:
        species = response.json()
        print(f"Species: {species['scientific_name']}")
        print(f"Common name: {species['common_name']}")
        print(f"Vector status: {species['vector_status']}")
    else:
        print("Species not found")

Get species information in Spanish:

response = await client.get(
    "http://localhost:8000/api/v1/species/anopheles-gambiae",
    params={"lang": "es"}
)

`get_vector_species_endpoint(request: Request, db: DBConnection = Depends(get_db), disease_id: str | None = Query(None), lang: str = Query(en))` `async` ¶

Retrieve mosquito species that are disease vectors, optionally filtered by disease.

This endpoint returns a list of mosquito species known to be vectors for various diseases. When a specific disease ID is provided, only species that are vectors for that particular disease are returned. Without a disease filter, all species marked as vectors (excluding 'None' and 'Unknown' status) are returned.

Parameters:

Name	Type	Description	Default
`request`	`Request`	The FastAPI request object used for URL construction.	required
`db`	`DBConnection`	LanceDB database connection for querying species data.	`Depends(get_db)`
`disease_id`	`str \| None`	Optional specific disease ID to filter vectors. If provided, only species that are vectors for this disease will be returned.	`Query(None)`
`lang`	`str`	Language code for response localization (e.g., 'en', 'es', 'ru').	`Query(en)`

Returns:

Type	Description
	list[SpeciesBase]: A list of mosquito species that are disease vectors, filtered by the specified criteria.

Example

Get all mosquito species that are disease vectors in English:

import httpx

async with httpx.AsyncClient() as client:
    response = await client.get(
        "http://localhost:8000/api/v1/vector-species",
        params={"lang": "en"}
    )
    vector_species = response.json()
    print(f"Found {len(vector_species)} vector species")
    for species in vector_species:
        print(f"- {species['scientific_name']}: {species['vector_status']}")

Get only species that are vectors for malaria:

# First, you'd need to know the disease ID for malaria
response = await client.get(
    "http://localhost:8000/api/v1/vector-species",
    params={"disease_id": "malaria", "lang": "en"}
)
malaria_vectors = response.json()

Get vector species in Russian:

response = await client.get(
    "http://localhost:8000/api/v1/vector-species",
    params={"lang": "ru"}
)

`get_region_cache(request: Request) -> dict[str, dict[str, str]]` ¶

Dependency for the regions translation cache.

Provides access to the cached region translations data, which maps region codes to translated names in multiple languages.

Parameters:

Name	Type	Description	Default
`request`	`Request`	The FastAPI request object containing app state.	required

Returns:

Type	Description
`dict[str, dict[str, str]]`	dict[str, dict[str, str]]: Dictionary mapping language codes to region translation dictionaries.

Example

@app.get("/regions/{language}") async def get_regions_by_language( regions: dict = Depends(get_region_cache), language: str = "en" ): return regions.get(language, {})

Информация о болезнях¶

API болезней управляет данными, связанными с болезнями, и предоставляет endpoints для получения информации о болезнях, переносимых комарами.

`backend.routers.diseases` ¶

Disease-related API endpoints for the CulicidaeLab server.

This module provides FastAPI router endpoints for retrieving information about vector-borne diseases and their associated arthropod vectors. The endpoints support multiple languages and provide both list and detail views of diseases.

The module includes the following endpoints: - GET /diseases: Retrieve a paginated list of diseases with optional search filtering - GET /diseases/{disease_id}: Retrieve detailed information for a specific disease - GET /diseases/{disease_id}/vectors: Retrieve vector species associated with a disease

All endpoints support internationalization and return data in the requested language.

`Disease` ¶

Complete disease model with unique identifier.

Used for responses that include the database ID of the disease.

`model_config = {'from_attributes': True}` `class-attribute` ¶

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

`Config` ¶

from_attributes = True class-attribute ¶

bool(x) -> bool

Returns True when the argument x is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.

`DiseaseListResponse` ¶

Response model for paginated disease lists.

Contains the total count and list of diseases for API responses.

`model_config = {}` `class-attribute` ¶

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

`SpeciesBase` ¶

Base model for species information.

Contains core species identification fields used across different species-related models.

`model_config = {}` `class-attribute` ¶

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

`get_disease_list_endpoint(request: Request, db: DBConnection = Depends(get_db), search: str | None = Query(None), limit: int = Query(50), lang: str = Query(en))` `async` ¶

Retrieve a list of vector-borne diseases, optionally filtered by a search term.

This endpoint provides a paginated list of diseases that can be filtered by name or description. Results are localized based on the requested language and support various query parameters for flexible disease discovery and analysis.

Parameters:

Name	Type	Description	Default
`request`	`Request`	The FastAPI request object containing client information and headers.	required
`db`	`DBConnection`	Database connection dependency for querying disease data from the LanceDB vector database.	`Depends(get_db)`
`search`	`str \| None`	Optional search term to filter diseases by name or description. Performs case-insensitive substring matching. If None, returns all diseases. Examples: "malaria", "fever", "mosquito".	`Query(None)`
`limit`	`int`	Maximum number of diseases to return per page (1-200). Defaults to 50. Use this parameter for pagination control when dealing with large result sets.	`Query(50)`
`lang`	`str`	Language code for response localization and internationalization. Supported languages: 'en' (English), 'es' (Spanish), 'fr' (French), 'pt' (Portuguese). Defaults to 'en' for English responses.	`Query(en)`

Returns:

Name	Type	Description
`DiseaseListResponse`		A structured response containing: - count (int): Total number of diseases matching the search criteria - diseases (list[Disease]): List of disease objects with basic information

Example

Basic disease list retrieval: GET /diseases?limit=10&lang=en

Search for specific diseases: GET /diseases?search=malaria&limit=5&lang=en

Multilingual support: GET /diseases?search=fiebre&lang=en

Response format: { "count": 2, "diseases": [ { "id": "malaria", "name": "Malaria", "description": "A life-threatening disease caused by Plasmodium parasites...", "vectors": ["Anopheles mosquito"] }, { "id": "dengue", "name": "Dengue Fever", "description": "A mosquito-borne viral infection causing flu-like illness...", "vectors": ["Aedes aegypti", "Aedes albopictus"] } ] }

`get_disease_detail_endpoint(disease_id: str, request: Request, db: DBConnection = Depends(get_db), lang: str = Query(en))` `async` ¶

Retrieve detailed information for a specific disease by ID.

This endpoint provides comprehensive information about a specific vector-borne disease, including its description, associated vectors, symptoms, and other relevant details. Results are localized based on the requested language.

Parameters:

Name	Type	Description	Default
`disease_id`	`str`	Unique identifier for the disease (e.g., 'malaria', 'dengue'). This is typically the disease name in lowercase with hyphens.	required
`request`	`Request`	The FastAPI request object containing client information.	required
`db`	`DBConnection`	Database connection dependency for querying disease data.	`Depends(get_db)`
`lang`	`str`	Language code for response localization (e.g., 'en', 'es', 'fr'). Defaults to 'en' for English.	`Query(en)`

Returns:

Name	Type	Description
`Disease`		A detailed disease object containing all available information about the specified disease, or None if the disease is not found.

Raises:

Type	Description
`HTTPException`	If the disease with the specified ID is not found (404 status code).

Example

GET /diseases/malaria?lang=en Response: { "id": "malaria", "name": "Malaria", "description": "A life-threatening disease caused by Plasmodium parasites...", "symptoms": ["Fever", "Chills", "Headache"], "vectors": ["Anopheles gambiae", "Anopheles funestus"], "regions": ["Sub-Saharan Africa", "Southeast Asia"], "transmission": "Bite of infected female Anopheles mosquitoes" }

`get_disease_vectors_endpoint(disease_id: str, request: Request, db: DBConnection = Depends(get_db), lang: str = Query(en))` `async` ¶

Retrieve vector species associated with a specific disease.

This endpoint returns a list of arthropod vectors (primarily mosquitoes, ticks, and flies) that are known to transmit the specified disease. This information is crucial for understanding disease transmission patterns and vector control strategies.

Parameters:

Name	Type	Description	Default
`disease_id`	`str`	Unique identifier for the disease (e.g., 'malaria', 'lyme-disease'). This is typically the disease name in lowercase with hyphens.	required
`request`	`Request`	The FastAPI request object containing client information.	required
`db`	`DBConnection`	Database connection dependency for querying species data.	`Depends(get_db)`
`lang`	`str`	Language code for response localization (e.g., 'en', 'es', 'fr'). Defaults to 'en' for English.	`Query(en)`

Returns:

Type	Description
	list[SpeciesBase]: A list of species objects that serve as vectors for the specified disease. Each species includes taxonomic information and habitat details.

Raises:

Type	Description
`HTTPException`	If the disease with the specified ID is not found (404 status code).

Example

GET /diseases/malaria/vectors?lang=en Response: [ { "id": "anopheles-gambiae", "scientific_name": "Anopheles gambiae", "common_name": "African malaria mosquito", "genus": "Anopheles", "family": "Culicidae", "habitat": "Tropical and subtropical regions", "distribution": "Sub-Saharan Africa" }, { "id": "anopheles-funestus", "scientific_name": "Anopheles funestus", "common_name": "Common malaria mosquito", "genus": "Anopheles", "family": "Culicidae", "habitat": "Rural and peri-urban areas", "distribution": "Africa south of the Sahara" } ]

Географические данные¶

Географический API обрабатывает данные на основе местоположения и предоставляет endpoints для получения географической информации и региональных данных.

`backend.routers.geo` ¶

Geographic API endpoints for the CulicidaeLab server.

This module provides FastAPI router endpoints for retrieving geographic data and spatial information related to arthropod vectors and vector-borne diseases. The endpoints support various layer types and filtering options for spatial analysis and mapping applications.

The module includes the following endpoints: - GET /geo/{layer_type}: Retrieve geographic features for a specific layer type with optional spatial, temporal, and species-based filtering

All endpoints return GeoJSON-compliant data structures suitable for mapping applications and geographic information systems (GIS). The endpoints support bounding box filtering, date range filtering, and species-specific queries.

`VALID_LAYER_TYPES = ['distribution', 'observations', 'modeled', 'breeding_sites']` `module-attribute` ¶

Built-in mutable sequence.

If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.

`GeoJSONFeatureCollection` ¶

GeoJSON FeatureCollection model.

Represents a collection of GeoJSON Features for batch operations and API responses containing multiple geographic features.

`model_config = {}` `class-attribute` ¶

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

`get_geographic_layer(layer_type: str = Path(PydanticUndefined), db: DBConnection = Depends(get_db), species: str | None = Query(None), bbox: str | None = Query(None), start_date: str | None = Query(None), end_date: str | None = Query(None))` `async` ¶

Retrieve geographic features for a specific layer type with optional filtering.

This endpoint provides access to various geographic data layers related to arthropod vectors and vector-borne diseases. Currently supports observation data with spatial, temporal, and species-based filtering capabilities.

The endpoint returns GeoJSON FeatureCollection data that can be directly consumed by mapping libraries and GIS applications for visualization and spatial analysis.

Parameters:

Name	Type	Description	Default
`layer_type`	`str`	The type of geographic layer to retrieve. Must be one of: 'distribution', 'observations', 'modeled', 'breeding_sites'.	`Path(PydanticUndefined)`
`db`	`DBConnection`	Database connection for querying geographic data.	`Depends(get_db)`
`species`	`str \| None`	Comma-separated list of species scientific names to filter by. If provided, only features for the specified species will be returned. Example: "Aedes aegypti,Anopheles gambiae,Culex quinquefasciatus".	`Query(None)`
`bbox`	`str \| None`	Bounding box filter in the format "min_lon,min_lat,max_lon,max_lat". Filters features to only include those within the specified geographic bounds. Example: "-122.4194,37.7749,-122.0808,37.9128" (San Francisco area).	`Query(None)`
`start_date`	`str \| None`	Start date for temporal filtering in YYYY-MM-DD format. Only features observed on or after this date will be included. Example: "2023-01-01".	`Query(None)`
`end_date`	`str \| None`	End date for temporal filtering in YYYY-MM-DD format. Only features observed on or before this date will be included. Example: "2023-12-31".	`Query(None)`

Returns:

Name	Type	Description
`GeoJSONFeatureCollection`		A GeoJSON FeatureCollection containing the requested geographic features. Each feature includes geometry (Point) and properties with observation metadata such as species information, observation dates, and location details.

Raises:

Type	Description
`HTTPException`	If layer_type is invalid (400) or if bbox format is incorrect (400).

Examples:

Basic usage - retrieve all observation features:

GET /geo/observations

Filter by species:

GET /geo/observations?species=Aedes%20aegypti,Anopheles%20gambiae

Filter by geographic bounds (San Francisco):

GET /geo/observations?bbox=-122.4194,37.7749,-122.0808,37.9128

Filter by date range:

GET /geo/observations?start_date=2023-01-01&end_date=2023-06-30

Combined filters - species and location:

GET /geo/observations?species=Aedes%20aegypti&bbox=-74.0,40.7,-71.0,45.0

Combined filters - all parameters:

GET /geo/observations?species=Culex%20quinquefasciatus
                    &bbox=-118.5,34.0,-118.1,34.3
                    &start_date=2023-03-01&end_date=2023-09-30

Сервисы предсказаний¶

API предсказаний предоставляет сервисы идентификации видов и предсказаний на основе машинного обучения.

`backend.routers.prediction` ¶

Prediction router for mosquito species identification.

This module provides FastAPI endpoints for predicting mosquito species from uploaded images using AI-powered classification. The router handles image validation, coordinates with the prediction service, and returns structured results.

Main Components

APIRouter instance configured for prediction endpoints
predict_species endpoint for species identification

The prediction system supports

Multiple image formats (JPEG, PNG, etc.)
Real-time species identification with confidence scores
Optional image saving for predicted results
Comprehensive error handling and logging

Example

from fastapi import FastAPI from backend.routers.prediction import router

app = FastAPI() app.include_router(router, prefix="/api/v1")

Now available at POST /api/v1/predict¶

`prediction_service = <backend.services.prediction_service.PredictionService object at 0x000001C3E7BE5B50>` `module-attribute` ¶

Service for mosquito species prediction using the CulicidaeLab serve API.

This class provides a high-level interface for species identification from images. It is optimized for production use, leveraging an efficient inference backend with automatic model caching to ensure low latency.

Attributes:

Name	Type	Description
`save_predicted_images_enabled`	`bool`	Whether to save predicted images.
`model_id`	`str`	The identifier for the machine learning model being used.

Example

service = PredictionService() result, error = await service.predict_species(image_data, "test.jpg")

`PredictionResult` ¶

Model for prediction results.

`model_config = {}` `class-attribute` ¶

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

`predict_species(file: UploadFile = File(PydanticUndefined)) -> PredictionResult` `async` ¶

Predict mosquito species from an uploaded image.

This endpoint accepts an image file and uses AI-powered classification to identify the mosquito species. The prediction includes confidence scores and optional species information.

Parameters:

Name	Type	Description	Default
`file`	`UploadFile`	The image file to analyze. Must be a valid image format (JPEG, PNG, etc.). The file is validated for content type and non-empty content.	`File(PydanticUndefined)`

Returns:

Name	Type	Description
`PredictionResult`	`PredictionResult`	A structured response containing: - id: Species identifier (lowercase with underscores) - scientific_name: The predicted species name - probabilities: Dictionary of species -> confidence scores - model_id: Identifier of the AI model used - confidence: Confidence score of the top prediction - image_url_species: URL to processed image (if saving enabled)

Raises:

Type	Description
`HTTPException`	If the file is not an image (400 Bad Request)
`HTTPException`	If the file is empty (400 Bad Request)
`HTTPException`	If prediction fails (500 Internal Server Error)

Example

import requests

Using curl command:¶

curl -X POST "http://localhost:8000/predict" >>> # -H "accept: application/json" >>> # -H "Content-Type: multipart/form-data" >>> # -F "file=@mosquito_image.jpg"¶

Using Python requests:¶

response = requests.post( ... "http://localhost:8000/predict", ... files={"file": open("mosquito_image.jpg", "rb")} ... ) result = response.json() print(f"Predicted species: {result['scientific_name']}") print(f"Confidence: {result['confidence']:.2%}")

Управление наблюдениями¶

API наблюдений обрабатывает данные наблюдений комаров, включая запись и получение записей наблюдений.

`backend.routers.observation` ¶

Observation Router Module for CulicidaeLab Server API.

This module provides FastAPI router endpoints for managing mosquito observation records. It handles creation and retrieval of observation data, including species identification, location information, and metadata.

The router integrates with the observation service layer to persist and retrieve data from the database while providing proper validation and error handling.

Typical usage example

from backend.routers.observation import router app.include_router(router, prefix="/api/v1")

Endpoints

POST /observations - Create a new observation record GET /observations - Retrieve observations with optional filtering

`Observation` ¶

Complete observation model with unique identifier.

Extends ObservationBase with system-generated fields for storing complete observation records.

`model_config = {}` `class-attribute` ¶

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

`ObservationListResponse` ¶

Response model for paginated observation lists.

Contains the total count and list of observations for API responses.

`model_config = {}` `class-attribute` ¶

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

`create_observation(observation: Observation) -> Observation` `async` ¶

Create a new mosquito observation record.

This endpoint accepts a complete observation record and stores it in the database. The observation should include species identification, location data, and metadata. If no user_id is provided, a UUID will be automatically generated.

Parameters:

Name	Type	Description	Default
`observation`	`Observation`	Complete observation data including species information, location coordinates, count, and optional metadata. Must conform to the Observation schema with all required fields validated.	required

Returns:

Name	Type	Description
`Observation`	`Observation`	The created observation record with assigned ID and any server-generated fields.

Raises:

Type	Description
`HTTPException`	If observation creation fails due to validation errors or database issues. Returns 500 status code for server errors.

Example

from backend.schemas.observation_schemas import Observation, Location observation_data = Observation( ... species_scientific_name="Aedes aegypti", ... count=5, ... location=Location(lat=40.7128, lng=-74.0060), ... observed_at="2024-01-15T10:30:00Z", ... notes="Found near standing water" ... ) result = await create_observation(observation_data) print(f"Created observation with ID: {result.id}")

`get_observations(species_id: str | None = None, limit: int = 100, offset: int = 0, user_id: str = 'default_user_id') -> ObservationListResponse` `async` ¶

Retrieve mosquito observations with optional filtering.

This endpoint returns a paginated list of observation records. Results can be filtered by species and user, with configurable pagination limits.

Parameters:

Name	Type	Description	Default
`species_id`	`str \| None`	Optional species identifier to filter observations. If None, returns observations for all species. Should be a valid species UUID or identifier from the database.	`None`
`limit`	`int`	Maximum number of observations to return in a single response. Must be between 1 and 1000. Defaults to 100. Larger values are automatically capped at 1000 for performance.	`100`
`offset`	`int`	Number of observations to skip for pagination. Must be non-negative. Defaults to 0. Use with limit for paginated results.	`0`
`user_id`	`str`	Identifier for the user whose observations to retrieve. Currently defaults to "default_user_id" but should be replaced with proper authentication in production.	`'default_user_id'`

Returns:

Name	Type	Description
`ObservationListResponse`	`ObservationListResponse`	Paginated response containing the total count of matching observations and the list of observation records. Each observation includes full species, location, and metadata.

Raises:

Type	Description
`HTTPException`	If observation retrieval fails due to database errors or invalid parameters. Returns 500 status code for server errors.

Example

Get first 50 observations for all species¶

response = await get_observations(limit=50) print(f"Total observations: {response.count}")

Get observations for a specific species with pagination¶

response = await get_observations( ... species_id="aedes-aegypti-uuid", ... limit=25, ... offset=25 ... ) for obs in response.observations: ... print(f"Species: {obs.species_scientific_name}")

Get observations for a specific user (when auth is implemented)¶

response = await get_observations( ... user_id="authenticated-user-id", ... limit=10 ... )

`get_observation_service()` `async` ¶

Get or initialize the global observation service instance.

This function implements a singleton pattern for the ObservationService, ensuring that only one instance exists and is properly initialized. If the service hasn't been created yet, it creates a new instance and initializes it.

Returns:

Name	Type	Description
`ObservationService`		The global observation service instance, initialized and ready for use.

Example

service = await get_observation_service() observations = await service.get_observations(limit=10)

Фильтрация данных¶

API фильтров предоставляет endpoints для получения опций фильтров и управления возможностями фильтрации данных.

`backend.routers.filters` ¶

Router module for filter-related API endpoints in CulicidaeLab.

This module provides FastAPI router endpoints for retrieving filter options used throughout the CulicidaeLab application. It handles requests for species, regions, and data source filtering options with multi-language support.

The module includes

Filter options endpoint for retrieving available filter criteria
Integration with caching dependencies for performance
Localization support for multiple languages

Typical usage example

from backend.routers.filters import router app.include_router(router)

Attributes:

Name	Type	Description
`router`	`APIRouter`	FastAPI APIRouter instance containing filter-related endpoints.

`FilterOptions` ¶

Container model for all available filter options.

Provides comprehensive filtering options including species, regions, and data sources for use in API endpoints that support filtering functionality.

`model_config = {}` `class-attribute` ¶

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

`get_filter_options_endpoint(lang: str = Query(en), species_names: list[str] = Depends(get_species_cache), region_translations: dict = Depends(get_region_cache), data_source_translations: dict = Depends(get_data_source_cache))` `async` ¶

Retrieves available filter options for species, regions, and data sources.

This endpoint provides comprehensive filtering options that can be used across the CulicidaeLab application to filter disease data by various criteria.

Parameters:

Name	Type	Description	Default
`lang`	`str`	Language code for response localization (e.g., 'en' for English, 'es' for Spanish). Defaults to 'en'.	`Query(en)`
`species_names`	`list[str]`	List of available mosquito species names for filtering.	`Depends(get_species_cache)`
`region_translations`	`dict`	Dictionary mapping region codes to translated names.	`Depends(get_region_cache)`
`data_source_translations`	`dict`	Dictionary mapping data source codes to translated names.	`Depends(get_data_source_cache)`

Returns:

Name	Type	Description
`FilterOptions`		A structured response containing: - species: List of available mosquito species with their display names - regions: List of geographical regions with translated names - data_sources: List of data sources with translated names

Example

Get filter options in English¶

response = await get_filter_options_endpoint("en", [...], {...}, {...}) print(response.species) # List of species options print(response.regions) # List of region options

Get filter options in Spanish¶

response = await get_filter_options_endpoint("es", [...], {...}, {...}) print(response.species) # Lista de opciones de especies

`get_data_source_cache(request: Request) -> dict[str, dict[str, str]]` ¶

Dependency for the data sources translation cache.

Provides access to the cached data source translations, which maps data source codes to translated names in multiple languages.

Parameters:

Name	Type	Description	Default
`request`	`Request`	The FastAPI request object containing app state.	required

Returns:

Type	Description
`dict[str, dict[str, str]]`	dict[str, dict[str, str]]: Dictionary mapping language codes to data source translation dictionaries.

Example

@app.get("/datasources/{language}") async def get_data_sources_by_language( sources: dict = Depends(get_data_source_cache), language: str = "en" ): return sources.get(language, {})

`get_region_cache(request: Request) -> dict[str, dict[str, str]]` ¶

Dependency for the regions translation cache.

Provides access to the cached region translations data, which maps region codes to translated names in multiple languages.

Parameters:

Name	Type	Description	Default
`request`	`Request`	The FastAPI request object containing app state.	required

Returns:

Type	Description
`dict[str, dict[str, str]]`	dict[str, dict[str, str]]: Dictionary mapping language codes to region translation dictionaries.

Example

@app.get("/regions/{language}") async def get_regions_by_language( regions: dict = Depends(get_region_cache), language: str = "en" ): return regions.get(language, {})

`get_species_cache(request: Request) -> list[str]` ¶

Dependency for the species names list cache.

Provides access to the cached list of all available species names for validation and autocomplete functionality.

Parameters:

Name	Type	Description	Default
`request`	`Request`	The FastAPI request object containing app state.	required

Returns:

Type	Description
`list[str]`	list[str]: List of all available species names in the system.

Example

@app.get("/species") async def get_all_species( species_list: list[str] = Depends(get_species_cache) ): return {"species": species_list}

Справочник API¶

Обзор¶

Базовый URL¶

Аутентификация¶

Формат ответа¶

Интерактивная документация API¶

Модули API¶

Основное приложение¶

backend.main ¶

Starts the server on http://localhost:8000¶

Or run with custom host/port¶

api_router_prefix = '' module-attribute ¶

backend_dir = 'C:\\Users\\lenova\\culicidaelab-server\\backend' module-attribute ¶

settings = AppSettings(APP_NAME='CulicidaeLab API', API_V1_STR='/api', DATABASE_PATH='backend/data/.lancedb', SAVE_PREDICTED_IMAGES='1', BACKEND_CORS_ORIGINS=['http://localhost:8765', 'http://127.0.0.1:8765']) module-attribute ¶

static_dir = 'C:\\Users\\lenova\\culicidaelab-server\\backend\\static' module-attribute ¶

lifespan(app: FastAPI) ¶

read_root() async ¶

health_check() async ¶

get_db() ¶

load_all_datasource_translations(db: object, supported_langs: list[str]) -> dict[str, dict[str, str]] ¶

load_all_region_translations(db: object, supported_langs: list[str]) -> dict[str, dict[str, str]] ¶

load_all_species_names(db: object) -> list[str] ¶

Управление видами¶

backend.routers.species ¶

SpeciesBase ¶

model_config = {} class-attribute ¶

SpeciesDetail ¶

model_config = {} class-attribute ¶

SpeciesListResponse ¶

model_config = {} class-attribute ¶

get_species_list_endpoint(request: Request, db: DBConnection = Depends(get_db), search: str | None = Query(None), limit: int = Query(50), lang: str = Query(en)) async ¶

get_species_detail_endpoint(species_id: str, request: Request, region_cache: dict[str, dict[str, str]] = Depends(get_region_cache), db: DBConnection = Depends(get_db), lang: str = Query(en)) async ¶

get_vector_species_endpoint(request: Request, db: DBConnection = Depends(get_db), disease_id: str | None = Query(None), lang: str = Query(en)) async ¶

get_region_cache(request: Request) -> dict[str, dict[str, str]] ¶

Информация о болезнях¶

backend.routers.diseases ¶

Disease ¶

model_config = {'from_attributes': True} class-attribute ¶

Config ¶

DiseaseListResponse ¶

model_config = {} class-attribute ¶

SpeciesBase ¶

model_config = {} class-attribute ¶

get_disease_list_endpoint(request: Request, db: DBConnection = Depends(get_db), search: str | None = Query(None), limit: int = Query(50), lang: str = Query(en)) async ¶

get_disease_detail_endpoint(disease_id: str, request: Request, db: DBConnection = Depends(get_db), lang: str = Query(en)) async ¶

get_disease_vectors_endpoint(disease_id: str, request: Request, db: DBConnection = Depends(get_db), lang: str = Query(en)) async ¶

Географические данные¶

backend.routers.geo ¶

VALID_LAYER_TYPES = ['distribution', 'observations', 'modeled', 'breeding_sites'] module-attribute ¶

GeoJSONFeatureCollection ¶

model_config = {} class-attribute ¶

get_geographic_layer(layer_type: str = Path(PydanticUndefined), db: DBConnection = Depends(get_db), species: str | None = Query(None), bbox: str | None = Query(None), start_date: str | None = Query(None), end_date: str | None = Query(None)) async ¶

Сервисы предсказаний¶

backend.routers.prediction ¶

Now available at POST /api/v1/predict¶

prediction_service = <backend.services.prediction_service.PredictionService object at 0x000001C3E7BE5B50> module-attribute ¶

PredictionResult ¶

model_config = {} class-attribute ¶

predict_species(file: UploadFile = File(PydanticUndefined)) -> PredictionResult async ¶

Using curl command:¶

curl -X POST "http://localhost:8000/predict" >>> # -H "accept: application/json" >>> # -H "Content-Type: multipart/form-data" >>> # -F "file=@mosquito_image.jpg"¶

Using Python requests:¶

Управление наблюдениями¶

backend.routers.observation ¶

Observation ¶

model_config = {} class-attribute ¶

ObservationListResponse ¶

model_config = {} class-attribute ¶

create_observation(observation: Observation) -> Observation async ¶

get_observations(species_id: str | None = None, limit: int = 100, offset: int = 0, user_id: str = 'default_user_id') -> ObservationListResponse async ¶

Get first 50 observations for all species¶

Get observations for a specific species with pagination¶

Get observations for a specific user (when auth is implemented)¶

get_observation_service() async ¶

Фильтрация данных¶

backend.routers.filters ¶

FilterOptions ¶

model_config = {} class-attribute ¶

get_filter_options_endpoint(lang: str = Query(en), species_names: list[str] = Depends(get_species_cache), region_translations: dict = Depends(get_region_cache), data_source_translations: dict = Depends(get_data_source_cache)) async ¶

Get filter options in English¶

`backend.main` ¶

Starts the server on http://localhost:8000 ¶

`api_router_prefix = ''` `module-attribute` ¶

`backend_dir = 'C:\\Users\\lenova\\culicidaelab-server\\backend'` `module-attribute` ¶

`settings = AppSettings(APP_NAME='CulicidaeLab API', API_V1_STR='/api', DATABASE_PATH='backend/data/.lancedb', SAVE_PREDICTED_IMAGES='1', BACKEND_CORS_ORIGINS=['http://localhost:8765', 'http://127.0.0.1:8765'])` `module-attribute` ¶

`static_dir = 'C:\\Users\\lenova\\culicidaelab-server\\backend\\static'` `module-attribute` ¶

`lifespan(app: FastAPI)` ¶

`read_root()` `async` ¶

`health_check()` `async` ¶

`get_db()` ¶

`load_all_datasource_translations(db: object, supported_langs: list[str]) -> dict[str, dict[str, str]]` ¶

`load_all_region_translations(db: object, supported_langs: list[str]) -> dict[str, dict[str, str]]` ¶

`load_all_species_names(db: object) -> list[str]` ¶

`backend.routers.species` ¶

`SpeciesBase` ¶

`model_config = {}` `class-attribute` ¶

`SpeciesDetail` ¶

`model_config = {}` `class-attribute` ¶

`SpeciesListResponse` ¶

`model_config = {}` `class-attribute` ¶

`get_species_list_endpoint(request: Request, db: DBConnection = Depends(get_db), search: str | None = Query(None), limit: int = Query(50), lang: str = Query(en))` `async` ¶

`get_species_detail_endpoint(species_id: str, request: Request, region_cache: dict[str, dict[str, str]] = Depends(get_region_cache), db: DBConnection = Depends(get_db), lang: str = Query(en))` `async` ¶

`get_vector_species_endpoint(request: Request, db: DBConnection = Depends(get_db), disease_id: str | None = Query(None), lang: str = Query(en))` `async` ¶

`get_region_cache(request: Request) -> dict[str, dict[str, str]]` ¶

`backend.routers.diseases` ¶

`Disease` ¶

`model_config = {'from_attributes': True}` `class-attribute` ¶

`Config` ¶

`DiseaseListResponse` ¶

`model_config = {}` `class-attribute` ¶

`SpeciesBase` ¶

`model_config = {}` `class-attribute` ¶

`get_disease_list_endpoint(request: Request, db: DBConnection = Depends(get_db), search: str | None = Query(None), limit: int = Query(50), lang: str = Query(en))` `async` ¶

`get_disease_detail_endpoint(disease_id: str, request: Request, db: DBConnection = Depends(get_db), lang: str = Query(en))` `async` ¶

`get_disease_vectors_endpoint(disease_id: str, request: Request, db: DBConnection = Depends(get_db), lang: str = Query(en))` `async` ¶

`backend.routers.geo` ¶

`VALID_LAYER_TYPES = ['distribution', 'observations', 'modeled', 'breeding_sites']` `module-attribute` ¶

`GeoJSONFeatureCollection` ¶

`model_config = {}` `class-attribute` ¶

`get_geographic_layer(layer_type: str = Path(PydanticUndefined), db: DBConnection = Depends(get_db), species: str | None = Query(None), bbox: str | None = Query(None), start_date: str | None = Query(None), end_date: str | None = Query(None))` `async` ¶

`backend.routers.prediction` ¶

`prediction_service = <backend.services.prediction_service.PredictionService object at 0x000001C3E7BE5B50>` `module-attribute` ¶

`PredictionResult` ¶

`model_config = {}` `class-attribute` ¶

`predict_species(file: UploadFile = File(PydanticUndefined)) -> PredictionResult` `async` ¶

`backend.routers.observation` ¶

`Observation` ¶

`model_config = {}` `class-attribute` ¶

`ObservationListResponse` ¶

`model_config = {}` `class-attribute` ¶

`create_observation(observation: Observation) -> Observation` `async` ¶

`get_observations(species_id: str | None = None, limit: int = 100, offset: int = 0, user_id: str = 'default_user_id') -> ObservationListResponse` `async` ¶

`get_observation_service()` `async` ¶

`backend.routers.filters` ¶

`FilterOptions` ¶

`model_config = {}` `class-attribute` ¶

`get_filter_options_endpoint(lang: str = Query(en), species_names: list[str] = Depends(get_species_cache), region_translations: dict = Depends(get_region_cache), data_source_translations: dict = Depends(get_data_source_cache))` `async` ¶

`get_data_source_cache(request: Request) -> dict[str, dict[str, str]]` ¶

`get_region_cache(request: Request) -> dict[str, dict[str, str]]` ¶

`get_species_cache(request: Request) -> list[str]` ¶