mspasspy.client

class mspasspy.client.Client(database_host=None, scheduler=None, scheduler_host=None, job_name='mspass', database_name='mspass', schema=None, collection=None, dask_client=None)

Bases: object

A client-side representation of MSPASS.

This is the only client users should use in MSPASS. The client manages all the other clients or instances. It creates and manages a Database client. It creates and manages a Global Hisotry Manager. It creates and manages a scheduler(spark/dask)

For the address and port of each client/instances, we first check the user specified parameters, if not then serach the environment varibales values, if not againm then use the default settings.

An existing dask.distributed.Client can be supplied with dask_client. This is useful for externally managed Dask clusters, including Dask Gateway clusters. The caller owns the external cluster and should keep it alive while the MsPASS client is using it.

close_scheduler()

Detach the active scheduler and release it when owned by this client.

A local Dask client or Spark context created by this Client is closed or stopped exactly once. A Dask client supplied by the caller and a Spark context reused from the caller are detached but remain running. Repeated calls are no-ops.

Scheduler state is cleared before an owned resource is released. If close or stop raises, that exception is propagated and this client remains detached from the scheduler.

get_database(database_name=None)

Get a database by database_name, if database_name is not specified, use the default one

Parameters:

database_name (str) – the name of database

Returns:

mspasspy.db.database.Database

get_database_client()

Get the database client in the global history manager

Returns:

mspasspy.db.database.Database

get_global_history_manager()

Get the global history manager with this client

Returns:

mspasspy.global_history.manager.GlobalHistoryManager

get_scheduler()

Get the scheduler(spark/dask) with this client

Returns:

pyspark.SparkContext/dask.distributed.Client/None

set_database_client(database_host, database_port=None)

Replace the database client and its global-history database together.

The replacement database client is connected and validated first. A new history database and GlobalHistoryManager are then built with the current history database name, schema objects, job name, and collection. Only after all of those steps succeed are the database and history references committed to this client. A failure therefore leaves the current database client, history manager, and scheduler unchanged.

An explicit port is appended only when database_host does not already contain one. This applies to bare hosts, MongoDB URIs, and bracketed IPv6 addresses.

Parameters:

• database_host (str) – the host address of database client

• database_port (str) – the port of database client

Raises:

MsPASSError – if the replacement database client, history database, or history manager cannot be constructed and validated.

set_global_history_manager(history_db, job_name, collection=None)

Set a global history manager by history_db, job_name(and collection)

Parameters:

• history_db (mspasspy.db.database.Database) – the database will be set in the global history manager

• job_name (str) – the job name will be set in the global history manager

• collection (str) – the collection name will be set in the history_db

set_scheduler(scheduler, scheduler_host, scheduler_port=None)

Replace the scheduler used by this client.

The replacement scheduler is constructed and validated before any scheduler state in this object is changed. Consequently, a failure to construct or validate the replacement leaves the current scheduler and its ownership unchanged. After a successful commit, a displaced Dask client or Spark context is closed only when it was created by this Client; a Dask client supplied by the caller is never closed.

Cleanup of an owned, displaced scheduler happens after the replacement is committed. If its close or stop method raises, that exception is propagated and the replacement remains the active scheduler. This distinction lets callers determine the active state with get_scheduler() without mistaking a cleanup failure for a failed connection.

Requesting the currently active Spark master is an idempotent no-op. Switching an active Spark context to another master is rejected because SparkSession.getOrCreate cannot guarantee that transition.

Parameters:

• scheduler (str) – the scheduler type, should be either dask or spark

• scheduler_host (str) – the host address of scheduler

• scheduler_port (str) – the port of scheduler

Raises:

• MsPASSError – if arguments are invalid, an optional scheduler dependency is unavailable, the replacement cannot be constructed or validated, or an active Spark master switch is requested.

• Exception – propagates an exception raised while closing or stopping a displaced Client-owned scheduler after the replacement has been committed.