test_rest_api

TestRestApiServer

TestRestApiServer(db_path=None, port=5000, host='127.0.0.1', prefix='', ssl=False, test=False, empty=False, models=False, logger=None)

A test REST API server that runs in a separate thread for testing purposes.

This class creates a Flask application that serves a PlantDB REST API and runs it in a separate thread so it doesn't block the current interpreter.

Attributes:

Name	Type	Description
db_path	Path	Path to the database directory
port	int	Port number for the server
host	str	Host address for the server
prefix	str	URL prefix for API endpoints
ssl	bool	Whether to use SSL/HTTPS
test	bool	Whether to run in test mode, that is to populate with the test dataset (default: False)
empty	bool	Whether to create an empty database (default: False)
models	bool	Whether to create a database with models (default: False)
app	Flask	Flask application instance
server	BaseWSGIServer	WSGI server instance
thread	Thread	Thread running the server

Examples:

>>> import requests
>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # EXAMPLE 1 - Create a test database and start the Flask App serving a REST API
>>> server = TestRestApiServer(test=True)
>>> server.start()
>>> # Get a list of all datasets from the DB:
>>> response = requests.get("http://127.0.0.1:5000/scans")
>>> scans_list = response.json()
>>> print(scans_list)
['arabidopsis000', 'real_plant', 'real_plant_analyzed', 'virtual_plant', 'virtual_plant_analyzed']
>>> server.stop()

>>> # EXAMPLE 2 - Serve an existing database
>>> from plantdb.commons.test_database import test_database
>>> test_db = test_database()  # set up a temporary test database
>>> print(test_db.path())
/tmp/ROMI_DB_********
>>> server = TestRestApiServer(db_path=test_db.path())
>>> server.start()
>>> # Get a list of all datasets from the DB:
>>> response = requests.get("http://127.0.0.1:5000/scans")
>>> scans_list = response.json()
>>> print(scans_list)
['real_plant_analyzed']
>>> server.stop()

Initialize the test REST API server.

Parameters:

Name	Type	Description	Default
db_path	str or Path	Path to the database directory to serve	None
port	int	Port number for the server (default: 5000)	5000
host	str	Host address for the server (default: '127.0.0.1')	'127.0.0.1'
prefix	str	URL prefix for API endpoints (default: '')	''
ssl	bool	Whether to use SSL/HTTPS (default: False)	False
test	bool	Whether to run in test mode, that is to populate with the test dataset (default: False)	False
empty	bool	Whether to create an empty database (default: False)	False
models	bool	Whether to create a database with models (default: False)	False
logger	logger	A logger instance to use. By default, the logger will be set to the root logger.	None

Returned by:

• Reference API server server test_rest_api  test_rest_api

Source code in plantdb/server/test_rest_api.py

def __init__(self, db_path=None, port=5000, host='127.0.0.1', prefix='', ssl=False, test=False, empty=False,
             models=False, logger=None):
    """Initialize the test REST API server.

    Parameters
    ----------
    db_path : str or pathlib.Path, optional
        Path to the database directory to serve
    port : int, optional
        Port number for the server (default: 5000)
    host : str, optional
        Host address for the server (default: '127.0.0.1')
    prefix : str, optional
        URL prefix for API endpoints (default: '')
    ssl : bool, optional
        Whether to use SSL/HTTPS (default: False)
    test : bool, optional
        Whether to run in test mode, that is to populate with the test dataset (default: False)
    empty : bool, optional
        Whether to create an empty database (default: False)
    models : bool, optional
        Whether to create a database with models (default: False)
    logger : logging.logger, optional
        A logger instance to use. By default, the logger will be set to the root logger.
    """
    self.db_path: Path = Path(db_path) if db_path else _mkdtemp_romidb()
    self.port: int = port
    self.host: str = host
    self.prefix: str = prefix
    self.ssl: bool = ssl
    self.test: bool = test
    self.empty: bool = empty
    self.models: bool = models
    self.app = None
    self.server = None
    self.thread = None
    self.logger: logging.Logger = logger if logger else get_logger(self.__class__.__name__) or get_logger(__name__)
    self._setup_flask_app()

__enter__

__enter__()

Context manager entry.

Source code in plantdb/server/test_rest_api.py

def __enter__(self):
    """Context manager entry."""
    self.start()
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)

Context manager exit.

Source code in plantdb/server/test_rest_api.py

def __exit__(self, exc_type, exc_val, exc_tb):
    """Context manager exit."""
    self.stop()

get_base_url

get_base_url()

Get the base URL for the API.

Returns:

Type	Description
str	The base URL for the API.

Source code in plantdb/server/test_rest_api.py

def get_base_url(self):
    """Get the base URL for the API.

    Returns
    -------
    str
        The base URL for the API.
    """
    protocol = 'https' if self.ssl else 'http'
    return f"{protocol}://{self.host}:{self.port}{self.prefix}"

get_server_config

get_server_config()

Get the server configuration settings.

Source code in plantdb/server/test_rest_api.py

def get_server_config(self):
    """Get the server configuration settings."""
    return {"host": self.host, "port": self.port, "prefix": self.prefix, "ssl": self.ssl}

is_running

is_running()

Check if the server is running.

Source code in plantdb/server/test_rest_api.py

def is_running(self):
    """Check if the server is running."""
    return self.thread is not None and self.thread.is_alive()

start

start()

Start the REST API server in a separate thread.

Source code in plantdb/server/test_rest_api.py

def start(self):
    """Start the REST API server in a separate thread."""
    if self.thread is not None and self.thread.is_alive():
        return  # Already running

    # Check if port is available
    if not self._is_port_available():
        raise RuntimeError(f"Port {self.port} is already in use on host {self.host}")

    # Create server
    self.server = make_server(self.host, self.port, self.app, threaded=True)

    # Start a server in a separate thread
    self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
    self.thread.start()

    # Wait a bit for the server to start
    time.sleep(0.5)

    self.logger.info(f"Test REST API server started at {self.get_base_url()}")

stop

stop()

Stop the REST API server.

Source code in plantdb/server/test_rest_api.py

def stop(self):
    """Stop the REST API server."""
    if self.server:
        self.server.shutdown()
        self.server = None

    if self.thread and self.thread.is_alive():
        self.thread.join(timeout=5.0)
        self.thread = None

    self.logger.info("Test REST API server stopped")

test_rest_api

test_rest_api(db_path, port=5000, host='127.0.0.1', prefix='', ssl=False)

Create and return a TestRestApi instance.

This is a convenience function that creates a TestRestApi instance with the specified parameters. The returned instance can be used to start/stop a test REST API server.

Parameters:

Name	Type	Description	Default
db_path	str or Path	The path to the database directory to serve	required
port	int	Port number for the server (default: 5000)	5000
host	str	Host address for the server (default: '127.0.0.1')	'127.0.0.1'
prefix	str	URL prefix for API endpoints (default: '/api/v1')	''
ssl	bool	Whether to use SSL/HTTPS (default: False)	False

Returns:

Type	Description
TestRestApiServer	Configured TestRestApi instance

Examples:

>>> from plantdb.commons.test_database import test_database
>>> from plantdb.server.test_rest_api import test_rest_api
>>> # Create a test database
>>> db = test_database(dataset=None)
>>> # Create and start a REST API server
>>> api = test_rest_api(db.path(), port=8080)
>>> api.start()
>>> # Use the API
>>> print(f"API running at: {api.get_base_url()}")
>>> # Stop the server
>>> api.stop()
>>>
>>> # Or use as a context manager
>>> with test_rest_api(db.path(), port=8080) as api:
...     # API is running here
...     print(f"API URL: {api.get_base_url()}")
>>> # API is automatically stopped here

Source code in plantdb/server/test_rest_api.py

def test_rest_api(db_path, port=5000, host='127.0.0.1', prefix='', ssl=False):
    """Create and return a TestRestApi instance.

    This is a convenience function that creates a TestRestApi instance with the
    specified parameters. The returned instance can be used to start/stop a
    test REST API server.

    Parameters
    ----------
    db_path : str or pathlib.Path
        The path to the database directory to serve
    port : int, optional
        Port number for the server (default: 5000)
    host : str, optional
        Host address for the server (default: '127.0.0.1')
    prefix : str, optional
        URL prefix for API endpoints (default: '/api/v1')
    ssl : bool, optional
        Whether to use SSL/HTTPS (default: False)

    Returns
    -------
    TestRestApiServer
        Configured TestRestApi instance

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> from plantdb.server.test_rest_api import test_rest_api
    >>> # Create a test database
    >>> db = test_database(dataset=None)
    >>> # Create and start a REST API server
    >>> api = test_rest_api(db.path(), port=8080)
    >>> api.start()
    >>> # Use the API
    >>> print(f"API running at: {api.get_base_url()}")
    >>> # Stop the server
    >>> api.stop()
    >>>
    >>> # Or use as a context manager
    >>> with test_rest_api(db.path(), port=8080) as api:
    ...     # API is running here
    ...     print(f"API URL: {api.get_base_url()}")
    >>> # API is automatically stopped here
    """
    return TestRestApiServer(db_path, port, host, prefix, ssl)