1804 lines
58 KiB
Python
1804 lines
58 KiB
Python
|
# Copyright 2015-present MongoDB, Inc.
|
||
|
#
|
||
|
# Licensed under the Apache License, Version 2.0 (the "License"); you
|
||
|
# may not use this file except in compliance with the License. You
|
||
|
# may obtain a copy of the License at
|
||
|
#
|
||
|
# http://www.apache.org/licenses/LICENSE-2.0
|
||
|
#
|
||
|
# Unless required by applicable law or agreed to in writing, software
|
||
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
||
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||
|
# implied. See the License for the specific language governing
|
||
|
# permissions and limitations under the License.
|
||
|
|
||
|
"""Tools to monitor driver events.
|
||
|
|
||
|
.. versionadded:: 3.1
|
||
|
|
||
|
.. attention:: Starting in PyMongo 3.11, the monitoring classes outlined below
|
||
|
are included in the PyMongo distribution under the
|
||
|
:mod:`~pymongo.event_loggers` submodule.
|
||
|
|
||
|
Use :func:`register` to register global listeners for specific events.
|
||
|
Listeners must inherit from one of the abstract classes below and implement
|
||
|
the correct functions for that class.
|
||
|
|
||
|
For example, a simple command logger might be implemented like this::
|
||
|
|
||
|
import logging
|
||
|
|
||
|
from pymongo import monitoring
|
||
|
|
||
|
class CommandLogger(monitoring.CommandListener):
|
||
|
|
||
|
def started(self, event):
|
||
|
logging.info("Command {0.command_name} with request id "
|
||
|
"{0.request_id} started on server "
|
||
|
"{0.connection_id}".format(event))
|
||
|
|
||
|
def succeeded(self, event):
|
||
|
logging.info("Command {0.command_name} with request id "
|
||
|
"{0.request_id} on server {0.connection_id} "
|
||
|
"succeeded in {0.duration_micros} "
|
||
|
"microseconds".format(event))
|
||
|
|
||
|
def failed(self, event):
|
||
|
logging.info("Command {0.command_name} with request id "
|
||
|
"{0.request_id} on server {0.connection_id} "
|
||
|
"failed in {0.duration_micros} "
|
||
|
"microseconds".format(event))
|
||
|
|
||
|
monitoring.register(CommandLogger())
|
||
|
|
||
|
Server discovery and monitoring events are also available. For example::
|
||
|
|
||
|
class ServerLogger(monitoring.ServerListener):
|
||
|
|
||
|
def opened(self, event):
|
||
|
logging.info("Server {0.server_address} added to topology "
|
||
|
"{0.topology_id}".format(event))
|
||
|
|
||
|
def description_changed(self, event):
|
||
|
previous_server_type = event.previous_description.server_type
|
||
|
new_server_type = event.new_description.server_type
|
||
|
if new_server_type != previous_server_type:
|
||
|
# server_type_name was added in PyMongo 3.4
|
||
|
logging.info(
|
||
|
"Server {0.server_address} changed type from "
|
||
|
"{0.previous_description.server_type_name} to "
|
||
|
"{0.new_description.server_type_name}".format(event))
|
||
|
|
||
|
def closed(self, event):
|
||
|
logging.warning("Server {0.server_address} removed from topology "
|
||
|
"{0.topology_id}".format(event))
|
||
|
|
||
|
|
||
|
class HeartbeatLogger(monitoring.ServerHeartbeatListener):
|
||
|
|
||
|
def started(self, event):
|
||
|
logging.info("Heartbeat sent to server "
|
||
|
"{0.connection_id}".format(event))
|
||
|
|
||
|
def succeeded(self, event):
|
||
|
# The reply.document attribute was added in PyMongo 3.4.
|
||
|
logging.info("Heartbeat to server {0.connection_id} "
|
||
|
"succeeded with reply "
|
||
|
"{0.reply.document}".format(event))
|
||
|
|
||
|
def failed(self, event):
|
||
|
logging.warning("Heartbeat to server {0.connection_id} "
|
||
|
"failed with error {0.reply}".format(event))
|
||
|
|
||
|
class TopologyLogger(monitoring.TopologyListener):
|
||
|
|
||
|
def opened(self, event):
|
||
|
logging.info("Topology with id {0.topology_id} "
|
||
|
"opened".format(event))
|
||
|
|
||
|
def description_changed(self, event):
|
||
|
logging.info("Topology description updated for "
|
||
|
"topology id {0.topology_id}".format(event))
|
||
|
previous_topology_type = event.previous_description.topology_type
|
||
|
new_topology_type = event.new_description.topology_type
|
||
|
if new_topology_type != previous_topology_type:
|
||
|
# topology_type_name was added in PyMongo 3.4
|
||
|
logging.info(
|
||
|
"Topology {0.topology_id} changed type from "
|
||
|
"{0.previous_description.topology_type_name} to "
|
||
|
"{0.new_description.topology_type_name}".format(event))
|
||
|
# The has_writable_server and has_readable_server methods
|
||
|
# were added in PyMongo 3.4.
|
||
|
if not event.new_description.has_writable_server():
|
||
|
logging.warning("No writable servers available.")
|
||
|
if not event.new_description.has_readable_server():
|
||
|
logging.warning("No readable servers available.")
|
||
|
|
||
|
def closed(self, event):
|
||
|
logging.info("Topology with id {0.topology_id} "
|
||
|
"closed".format(event))
|
||
|
|
||
|
Connection monitoring and pooling events are also available. For example::
|
||
|
|
||
|
class ConnectionPoolLogger(ConnectionPoolListener):
|
||
|
|
||
|
def pool_created(self, event):
|
||
|
logging.info("[pool {0.address}] pool created".format(event))
|
||
|
|
||
|
def pool_ready(self, event):
|
||
|
logging.info("[pool {0.address}] pool is ready".format(event))
|
||
|
|
||
|
def pool_cleared(self, event):
|
||
|
logging.info("[pool {0.address}] pool cleared".format(event))
|
||
|
|
||
|
def pool_closed(self, event):
|
||
|
logging.info("[pool {0.address}] pool closed".format(event))
|
||
|
|
||
|
def connection_created(self, event):
|
||
|
logging.info("[pool {0.address}][conn #{0.connection_id}] "
|
||
|
"connection created".format(event))
|
||
|
|
||
|
def connection_ready(self, event):
|
||
|
logging.info("[pool {0.address}][conn #{0.connection_id}] "
|
||
|
"connection setup succeeded".format(event))
|
||
|
|
||
|
def connection_closed(self, event):
|
||
|
logging.info("[pool {0.address}][conn #{0.connection_id}] "
|
||
|
"connection closed, reason: "
|
||
|
"{0.reason}".format(event))
|
||
|
|
||
|
def connection_check_out_started(self, event):
|
||
|
logging.info("[pool {0.address}] connection check out "
|
||
|
"started".format(event))
|
||
|
|
||
|
def connection_check_out_failed(self, event):
|
||
|
logging.info("[pool {0.address}] connection check out "
|
||
|
"failed, reason: {0.reason}".format(event))
|
||
|
|
||
|
def connection_checked_out(self, event):
|
||
|
logging.info("[pool {0.address}][conn #{0.connection_id}] "
|
||
|
"connection checked out of pool".format(event))
|
||
|
|
||
|
def connection_checked_in(self, event):
|
||
|
logging.info("[pool {0.address}][conn #{0.connection_id}] "
|
||
|
"connection checked into pool".format(event))
|
||
|
|
||
|
|
||
|
Event listeners can also be registered per instance of
|
||
|
:class:`~pymongo.mongo_client.MongoClient`::
|
||
|
|
||
|
client = MongoClient(event_listeners=[CommandLogger()])
|
||
|
|
||
|
Note that previously registered global listeners are automatically included
|
||
|
when configuring per client event listeners. Registering a new global listener
|
||
|
will not add that listener to existing client instances.
|
||
|
|
||
|
.. note:: Events are delivered **synchronously**. Application threads block
|
||
|
waiting for event handlers (e.g. :meth:`~CommandListener.started`) to
|
||
|
return. Care must be taken to ensure that your event handlers are efficient
|
||
|
enough to not adversely affect overall application performance.
|
||
|
|
||
|
.. warning:: The command documents published through this API are *not* copies.
|
||
|
If you intend to modify them in any way you must copy them in your event
|
||
|
handler first.
|
||
|
"""
|
||
|
|
||
|
from __future__ import annotations
|
||
|
|
||
|
import datetime
|
||
|
from collections import abc, namedtuple
|
||
|
from typing import TYPE_CHECKING, Any, Dict, Optional
|
||
|
|
||
|
from bson.objectid import ObjectId
|
||
|
from pymongo.hello import Hello, HelloCompat
|
||
|
from pymongo.helpers import _handle_exception
|
||
|
from pymongo.typings import _Address, _DocumentOut
|
||
|
|
||
|
if TYPE_CHECKING:
|
||
|
from pymongo.server_description import ServerDescription
|
||
|
from pymongo.topology_description import TopologyDescription
|
||
|
|
||
|
|
||
|
_Listeners = namedtuple(
|
||
|
"_Listeners",
|
||
|
(
|
||
|
"command_listeners",
|
||
|
"server_listeners",
|
||
|
"server_heartbeat_listeners",
|
||
|
"topology_listeners",
|
||
|
"cmap_listeners",
|
||
|
),
|
||
|
)
|
||
|
|
||
|
_LISTENERS = _Listeners([], [], [], [], [])
|
||
|
|
||
|
|
||
|
class _EventListener:
|
||
|
"""Abstract base class for all event listeners."""
|
||
|
|
||
|
|
||
|
class CommandListener(_EventListener):
|
||
|
"""Abstract base class for command listeners.
|
||
|
|
||
|
Handles `CommandStartedEvent`, `CommandSucceededEvent`,
|
||
|
and `CommandFailedEvent`.
|
||
|
"""
|
||
|
|
||
|
def started(self, event: "CommandStartedEvent") -> None:
|
||
|
"""Abstract method to handle a `CommandStartedEvent`.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`CommandStartedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def succeeded(self, event: "CommandSucceededEvent") -> None:
|
||
|
"""Abstract method to handle a `CommandSucceededEvent`.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`CommandSucceededEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def failed(self, event: "CommandFailedEvent") -> None:
|
||
|
"""Abstract method to handle a `CommandFailedEvent`.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`CommandFailedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
|
||
|
class ConnectionPoolListener(_EventListener):
|
||
|
"""Abstract base class for connection pool listeners.
|
||
|
|
||
|
Handles all of the connection pool events defined in the Connection
|
||
|
Monitoring and Pooling Specification:
|
||
|
:class:`PoolCreatedEvent`, :class:`PoolClearedEvent`,
|
||
|
:class:`PoolClosedEvent`, :class:`ConnectionCreatedEvent`,
|
||
|
:class:`ConnectionReadyEvent`, :class:`ConnectionClosedEvent`,
|
||
|
:class:`ConnectionCheckOutStartedEvent`,
|
||
|
:class:`ConnectionCheckOutFailedEvent`,
|
||
|
:class:`ConnectionCheckedOutEvent`,
|
||
|
and :class:`ConnectionCheckedInEvent`.
|
||
|
|
||
|
.. versionadded:: 3.9
|
||
|
"""
|
||
|
|
||
|
def pool_created(self, event: "PoolCreatedEvent") -> None:
|
||
|
"""Abstract method to handle a :class:`PoolCreatedEvent`.
|
||
|
|
||
|
Emitted when a Connection Pool is created.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`PoolCreatedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def pool_ready(self, event: "PoolReadyEvent") -> None:
|
||
|
"""Abstract method to handle a :class:`PoolReadyEvent`.
|
||
|
|
||
|
Emitted when a Connection Pool is marked ready.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`PoolReadyEvent`.
|
||
|
|
||
|
.. versionadded:: 4.0
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def pool_cleared(self, event: "PoolClearedEvent") -> None:
|
||
|
"""Abstract method to handle a `PoolClearedEvent`.
|
||
|
|
||
|
Emitted when a Connection Pool is cleared.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`PoolClearedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def pool_closed(self, event: "PoolClosedEvent") -> None:
|
||
|
"""Abstract method to handle a `PoolClosedEvent`.
|
||
|
|
||
|
Emitted when a Connection Pool is closed.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`PoolClosedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def connection_created(self, event: "ConnectionCreatedEvent") -> None:
|
||
|
"""Abstract method to handle a :class:`ConnectionCreatedEvent`.
|
||
|
|
||
|
Emitted when a Connection Pool creates a Connection object.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`ConnectionCreatedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def connection_ready(self, event: "ConnectionReadyEvent") -> None:
|
||
|
"""Abstract method to handle a :class:`ConnectionReadyEvent`.
|
||
|
|
||
|
Emitted when a Connection has finished its setup, and is now ready to
|
||
|
use.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`ConnectionReadyEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def connection_closed(self, event: "ConnectionClosedEvent") -> None:
|
||
|
"""Abstract method to handle a :class:`ConnectionClosedEvent`.
|
||
|
|
||
|
Emitted when a Connection Pool closes a Connection.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`ConnectionClosedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def connection_check_out_started(self, event: "ConnectionCheckOutStartedEvent") -> None:
|
||
|
"""Abstract method to handle a :class:`ConnectionCheckOutStartedEvent`.
|
||
|
|
||
|
Emitted when the driver starts attempting to check out a connection.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`ConnectionCheckOutStartedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def connection_check_out_failed(self, event: "ConnectionCheckOutFailedEvent") -> None:
|
||
|
"""Abstract method to handle a :class:`ConnectionCheckOutFailedEvent`.
|
||
|
|
||
|
Emitted when the driver's attempt to check out a connection fails.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`ConnectionCheckOutFailedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def connection_checked_out(self, event: "ConnectionCheckedOutEvent") -> None:
|
||
|
"""Abstract method to handle a :class:`ConnectionCheckedOutEvent`.
|
||
|
|
||
|
Emitted when the driver successfully checks out a Connection.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`ConnectionCheckedOutEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def connection_checked_in(self, event: "ConnectionCheckedInEvent") -> None:
|
||
|
"""Abstract method to handle a :class:`ConnectionCheckedInEvent`.
|
||
|
|
||
|
Emitted when the driver checks in a Connection back to the Connection
|
||
|
Pool.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`ConnectionCheckedInEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
|
||
|
class ServerHeartbeatListener(_EventListener):
|
||
|
"""Abstract base class for server heartbeat listeners.
|
||
|
|
||
|
Handles `ServerHeartbeatStartedEvent`, `ServerHeartbeatSucceededEvent`,
|
||
|
and `ServerHeartbeatFailedEvent`.
|
||
|
|
||
|
.. versionadded:: 3.3
|
||
|
"""
|
||
|
|
||
|
def started(self, event: "ServerHeartbeatStartedEvent") -> None:
|
||
|
"""Abstract method to handle a `ServerHeartbeatStartedEvent`.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`ServerHeartbeatStartedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def succeeded(self, event: "ServerHeartbeatSucceededEvent") -> None:
|
||
|
"""Abstract method to handle a `ServerHeartbeatSucceededEvent`.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`ServerHeartbeatSucceededEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def failed(self, event: "ServerHeartbeatFailedEvent") -> None:
|
||
|
"""Abstract method to handle a `ServerHeartbeatFailedEvent`.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`ServerHeartbeatFailedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
|
||
|
class TopologyListener(_EventListener):
|
||
|
"""Abstract base class for topology monitoring listeners.
|
||
|
Handles `TopologyOpenedEvent`, `TopologyDescriptionChangedEvent`, and
|
||
|
`TopologyClosedEvent`.
|
||
|
|
||
|
.. versionadded:: 3.3
|
||
|
"""
|
||
|
|
||
|
def opened(self, event: "TopologyOpenedEvent") -> None:
|
||
|
"""Abstract method to handle a `TopologyOpenedEvent`.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`TopologyOpenedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def description_changed(self, event: "TopologyDescriptionChangedEvent") -> None:
|
||
|
"""Abstract method to handle a `TopologyDescriptionChangedEvent`.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`TopologyDescriptionChangedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def closed(self, event: "TopologyClosedEvent") -> None:
|
||
|
"""Abstract method to handle a `TopologyClosedEvent`.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`TopologyClosedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
|
||
|
class ServerListener(_EventListener):
|
||
|
"""Abstract base class for server listeners.
|
||
|
Handles `ServerOpeningEvent`, `ServerDescriptionChangedEvent`, and
|
||
|
`ServerClosedEvent`.
|
||
|
|
||
|
.. versionadded:: 3.3
|
||
|
"""
|
||
|
|
||
|
def opened(self, event: "ServerOpeningEvent") -> None:
|
||
|
"""Abstract method to handle a `ServerOpeningEvent`.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`ServerOpeningEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def description_changed(self, event: "ServerDescriptionChangedEvent") -> None:
|
||
|
"""Abstract method to handle a `ServerDescriptionChangedEvent`.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`ServerDescriptionChangedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
def closed(self, event: "ServerClosedEvent") -> None:
|
||
|
"""Abstract method to handle a `ServerClosedEvent`.
|
||
|
|
||
|
:Parameters:
|
||
|
- `event`: An instance of :class:`ServerClosedEvent`.
|
||
|
"""
|
||
|
raise NotImplementedError
|
||
|
|
||
|
|
||
|
def _to_micros(dur):
|
||
|
"""Convert duration 'dur' to microseconds."""
|
||
|
return int(dur.total_seconds() * 10e5)
|
||
|
|
||
|
|
||
|
def _validate_event_listeners(option, listeners):
|
||
|
"""Validate event listeners"""
|
||
|
if not isinstance(listeners, abc.Sequence):
|
||
|
raise TypeError(f"{option} must be a list or tuple")
|
||
|
for listener in listeners:
|
||
|
if not isinstance(listener, _EventListener):
|
||
|
raise TypeError(
|
||
|
"Listeners for {} must be either a "
|
||
|
"CommandListener, ServerHeartbeatListener, "
|
||
|
"ServerListener, TopologyListener, or "
|
||
|
"ConnectionPoolListener.".format(option)
|
||
|
)
|
||
|
return listeners
|
||
|
|
||
|
|
||
|
def register(listener: _EventListener) -> None:
|
||
|
"""Register a global event listener.
|
||
|
|
||
|
:Parameters:
|
||
|
- `listener`: A subclasses of :class:`CommandListener`,
|
||
|
:class:`ServerHeartbeatListener`, :class:`ServerListener`,
|
||
|
:class:`TopologyListener`, or :class:`ConnectionPoolListener`.
|
||
|
"""
|
||
|
if not isinstance(listener, _EventListener):
|
||
|
raise TypeError(
|
||
|
"Listeners for {} must be either a "
|
||
|
"CommandListener, ServerHeartbeatListener, "
|
||
|
"ServerListener, TopologyListener, or "
|
||
|
"ConnectionPoolListener.".format(listener)
|
||
|
)
|
||
|
if isinstance(listener, CommandListener):
|
||
|
_LISTENERS.command_listeners.append(listener)
|
||
|
if isinstance(listener, ServerHeartbeatListener):
|
||
|
_LISTENERS.server_heartbeat_listeners.append(listener)
|
||
|
if isinstance(listener, ServerListener):
|
||
|
_LISTENERS.server_listeners.append(listener)
|
||
|
if isinstance(listener, TopologyListener):
|
||
|
_LISTENERS.topology_listeners.append(listener)
|
||
|
if isinstance(listener, ConnectionPoolListener):
|
||
|
_LISTENERS.cmap_listeners.append(listener)
|
||
|
|
||
|
|
||
|
# Note - to avoid bugs from forgetting which if these is all lowercase and
|
||
|
# which are camelCase, and at the same time avoid having to add a test for
|
||
|
# every command, use all lowercase here and test against command_name.lower().
|
||
|
_SENSITIVE_COMMANDS: set = {
|
||
|
"authenticate",
|
||
|
"saslstart",
|
||
|
"saslcontinue",
|
||
|
"getnonce",
|
||
|
"createuser",
|
||
|
"updateuser",
|
||
|
"copydbgetnonce",
|
||
|
"copydbsaslstart",
|
||
|
"copydb",
|
||
|
}
|
||
|
|
||
|
|
||
|
# The "hello" command is also deemed sensitive when attempting speculative
|
||
|
# authentication.
|
||
|
def _is_speculative_authenticate(command_name, doc):
|
||
|
if (
|
||
|
command_name.lower() in ("hello", HelloCompat.LEGACY_CMD)
|
||
|
and "speculativeAuthenticate" in doc
|
||
|
):
|
||
|
return True
|
||
|
return False
|
||
|
|
||
|
|
||
|
class _CommandEvent:
|
||
|
"""Base class for command events."""
|
||
|
|
||
|
__slots__ = ("__cmd_name", "__rqst_id", "__conn_id", "__op_id", "__service_id")
|
||
|
|
||
|
def __init__(
|
||
|
self,
|
||
|
command_name: str,
|
||
|
request_id: int,
|
||
|
connection_id: _Address,
|
||
|
operation_id: Optional[int],
|
||
|
service_id: Optional[ObjectId] = None,
|
||
|
) -> None:
|
||
|
self.__cmd_name = command_name
|
||
|
self.__rqst_id = request_id
|
||
|
self.__conn_id = connection_id
|
||
|
self.__op_id = operation_id
|
||
|
self.__service_id = service_id
|
||
|
|
||
|
@property
|
||
|
def command_name(self) -> str:
|
||
|
"""The command name."""
|
||
|
return self.__cmd_name
|
||
|
|
||
|
@property
|
||
|
def request_id(self) -> int:
|
||
|
"""The request id for this operation."""
|
||
|
return self.__rqst_id
|
||
|
|
||
|
@property
|
||
|
def connection_id(self) -> _Address:
|
||
|
"""The address (host, port) of the server this command was sent to."""
|
||
|
return self.__conn_id
|
||
|
|
||
|
@property
|
||
|
def service_id(self) -> Optional[ObjectId]:
|
||
|
"""The service_id this command was sent to, or ``None``.
|
||
|
|
||
|
.. versionadded:: 3.12
|
||
|
"""
|
||
|
return self.__service_id
|
||
|
|
||
|
@property
|
||
|
def operation_id(self) -> Optional[int]:
|
||
|
"""An id for this series of events or None."""
|
||
|
return self.__op_id
|
||
|
|
||
|
|
||
|
class CommandStartedEvent(_CommandEvent):
|
||
|
"""Event published when a command starts.
|
||
|
|
||
|
:Parameters:
|
||
|
- `command`: The command document.
|
||
|
- `database_name`: The name of the database this command was run against.
|
||
|
- `request_id`: The request id for this operation.
|
||
|
- `connection_id`: The address (host, port) of the server this command
|
||
|
was sent to.
|
||
|
- `operation_id`: An optional identifier for a series of related events.
|
||
|
- `service_id`: The service_id this command was sent to, or ``None``.
|
||
|
"""
|
||
|
|
||
|
__slots__ = ("__cmd", "__db")
|
||
|
|
||
|
def __init__(
|
||
|
self,
|
||
|
command: _DocumentOut,
|
||
|
database_name: str,
|
||
|
request_id: int,
|
||
|
connection_id: _Address,
|
||
|
operation_id: Optional[int],
|
||
|
service_id: Optional[ObjectId] = None,
|
||
|
) -> None:
|
||
|
if not command:
|
||
|
raise ValueError(f"{command!r} is not a valid command")
|
||
|
# Command name must be first key.
|
||
|
command_name = next(iter(command))
|
||
|
super().__init__(
|
||
|
command_name, request_id, connection_id, operation_id, service_id=service_id
|
||
|
)
|
||
|
cmd_name = command_name.lower()
|
||
|
if cmd_name in _SENSITIVE_COMMANDS or _is_speculative_authenticate(cmd_name, command):
|
||
|
self.__cmd: _DocumentOut = {}
|
||
|
else:
|
||
|
self.__cmd = command
|
||
|
self.__db = database_name
|
||
|
|
||
|
@property
|
||
|
def command(self) -> _DocumentOut:
|
||
|
"""The command document."""
|
||
|
return self.__cmd
|
||
|
|
||
|
@property
|
||
|
def database_name(self) -> str:
|
||
|
"""The name of the database this command was run against."""
|
||
|
return self.__db
|
||
|
|
||
|
def __repr__(self):
|
||
|
return ("<{} {} db: {!r}, command: {!r}, operation_id: {}, service_id: {}>").format(
|
||
|
self.__class__.__name__,
|
||
|
self.connection_id,
|
||
|
self.database_name,
|
||
|
self.command_name,
|
||
|
self.operation_id,
|
||
|
self.service_id,
|
||
|
)
|
||
|
|
||
|
|
||
|
class CommandSucceededEvent(_CommandEvent):
|
||
|
"""Event published when a command succeeds.
|
||
|
|
||
|
:Parameters:
|
||
|
- `duration`: The command duration as a datetime.timedelta.
|
||
|
- `reply`: The server reply document.
|
||
|
- `command_name`: The command name.
|
||
|
- `request_id`: The request id for this operation.
|
||
|
- `connection_id`: The address (host, port) of the server this command
|
||
|
was sent to.
|
||
|
- `operation_id`: An optional identifier for a series of related events.
|
||
|
- `service_id`: The service_id this command was sent to, or ``None``.
|
||
|
"""
|
||
|
|
||
|
__slots__ = ("__duration_micros", "__reply")
|
||
|
|
||
|
def __init__(
|
||
|
self,
|
||
|
duration: datetime.timedelta,
|
||
|
reply: _DocumentOut,
|
||
|
command_name: str,
|
||
|
request_id: int,
|
||
|
connection_id: _Address,
|
||
|
operation_id: Optional[int],
|
||
|
service_id: Optional[ObjectId] = None,
|
||
|
) -> None:
|
||
|
super().__init__(
|
||
|
command_name, request_id, connection_id, operation_id, service_id=service_id
|
||
|
)
|
||
|
self.__duration_micros = _to_micros(duration)
|
||
|
cmd_name = command_name.lower()
|
||
|
if cmd_name in _SENSITIVE_COMMANDS or _is_speculative_authenticate(cmd_name, reply):
|
||
|
self.__reply: _DocumentOut = {}
|
||
|
else:
|
||
|
self.__reply = reply
|
||
|
|
||
|
@property
|
||
|
def duration_micros(self) -> int:
|
||
|
"""The duration of this operation in microseconds."""
|
||
|
return self.__duration_micros
|
||
|
|
||
|
@property
|
||
|
def reply(self) -> _DocumentOut:
|
||
|
"""The server failure document for this operation."""
|
||
|
return self.__reply
|
||
|
|
||
|
def __repr__(self):
|
||
|
return (
|
||
|
"<{} {} command: {!r}, operation_id: {}, duration_micros: {}, service_id: {}>"
|
||
|
).format(
|
||
|
self.__class__.__name__,
|
||
|
self.connection_id,
|
||
|
self.command_name,
|
||
|
self.operation_id,
|
||
|
self.duration_micros,
|
||
|
self.service_id,
|
||
|
)
|
||
|
|
||
|
|
||
|
class CommandFailedEvent(_CommandEvent):
|
||
|
"""Event published when a command fails.
|
||
|
|
||
|
:Parameters:
|
||
|
- `duration`: The command duration as a datetime.timedelta.
|
||
|
- `failure`: The server reply document.
|
||
|
- `command_name`: The command name.
|
||
|
- `request_id`: The request id for this operation.
|
||
|
- `connection_id`: The address (host, port) of the server this command
|
||
|
was sent to.
|
||
|
- `operation_id`: An optional identifier for a series of related events.
|
||
|
- `service_id`: The service_id this command was sent to, or ``None``.
|
||
|
"""
|
||
|
|
||
|
__slots__ = ("__duration_micros", "__failure")
|
||
|
|
||
|
def __init__(
|
||
|
self,
|
||
|
duration: datetime.timedelta,
|
||
|
failure: _DocumentOut,
|
||
|
command_name: str,
|
||
|
request_id: int,
|
||
|
connection_id: _Address,
|
||
|
operation_id: Optional[int],
|
||
|
service_id: Optional[ObjectId] = None,
|
||
|
) -> None:
|
||
|
super().__init__(
|
||
|
command_name, request_id, connection_id, operation_id, service_id=service_id
|
||
|
)
|
||
|
self.__duration_micros = _to_micros(duration)
|
||
|
self.__failure = failure
|
||
|
|
||
|
@property
|
||
|
def duration_micros(self) -> int:
|
||
|
"""The duration of this operation in microseconds."""
|
||
|
return self.__duration_micros
|
||
|
|
||
|
@property
|
||
|
def failure(self) -> _DocumentOut:
|
||
|
"""The server failure document for this operation."""
|
||
|
return self.__failure
|
||
|
|
||
|
def __repr__(self):
|
||
|
return (
|
||
|
"<{} {} command: {!r}, operation_id: {}, duration_micros: {}, "
|
||
|
"failure: {!r}, service_id: {}>"
|
||
|
).format(
|
||
|
self.__class__.__name__,
|
||
|
self.connection_id,
|
||
|
self.command_name,
|
||
|
self.operation_id,
|
||
|
self.duration_micros,
|
||
|
self.failure,
|
||
|
self.service_id,
|
||
|
)
|
||
|
|
||
|
|
||
|
class _PoolEvent:
|
||
|
"""Base class for pool events."""
|
||
|
|
||
|
__slots__ = ("__address",)
|
||
|
|
||
|
def __init__(self, address: _Address) -> None:
|
||
|
self.__address = address
|
||
|
|
||
|
@property
|
||
|
def address(self) -> _Address:
|
||
|
"""The address (host, port) pair of the server the pool is attempting
|
||
|
to connect to.
|
||
|
"""
|
||
|
return self.__address
|
||
|
|
||
|
def __repr__(self):
|
||
|
return f"{self.__class__.__name__}({self.__address!r})"
|
||
|
|
||
|
|
||
|
class PoolCreatedEvent(_PoolEvent):
|
||
|
"""Published when a Connection Pool is created.
|
||
|
|
||
|
:Parameters:
|
||
|
- `address`: The address (host, port) pair of the server this Pool is
|
||
|
attempting to connect to.
|
||
|
|
||
|
.. versionadded:: 3.9
|
||
|
"""
|
||
|
|
||
|
__slots__ = ("__options",)
|
||
|
|
||
|
def __init__(self, address: _Address, options: Dict[str, Any]) -> None:
|
||
|
super().__init__(address)
|
||
|
self.__options = options
|
||
|
|
||
|
@property
|
||
|
def options(self) -> Dict[str, Any]:
|
||
|
"""Any non-default pool options that were set on this Connection Pool."""
|
||
|
return self.__options
|
||
|
|
||
|
def __repr__(self):
|
||
|
return f"{self.__class__.__name__}({self.address!r}, {self.__options!r})"
|
||
|
|
||
|
|
||
|
class PoolReadyEvent(_PoolEvent):
|
||
|
"""Published when a Connection Pool is marked ready.
|
||
|
|
||
|
:Parameters:
|
||
|
- `address`: The address (host, port) pair of the server this Pool is
|
||
|
attempting to connect to.
|
||
|
|
||
|
.. versionadded:: 4.0
|
||
|
"""
|
||
|
|
||
|
__slots__ = ()
|
||
|
|
||
|
|
||
|
class PoolClearedEvent(_PoolEvent):
|
||
|
"""Published when a Connection Pool is cleared.
|
||
|
|
||
|
:Parameters:
|
||
|
- `address`: The address (host, port) pair of the server this Pool is
|
||
|
attempting to connect to.
|
||
|
- `service_id`: The service_id this command was sent to, or ``None``.
|
||
|
|
||
|
.. versionadded:: 3.9
|
||
|
"""
|
||
|
|
||
|
__slots__ = ("__service_id",)
|
||
|
|
||
|
def __init__(self, address: _Address, service_id: Optional[ObjectId] = None) -> None:
|
||
|
super().__init__(address)
|
||
|
self.__service_id = service_id
|
||
|
|
||
|
@property
|
||
|
def service_id(self) -> Optional[ObjectId]:
|
||
|
"""Connections with this service_id are cleared.
|
||
|
|
||
|
When service_id is ``None``, all connections in the pool are cleared.
|
||
|
|
||
|
.. versionadded:: 3.12
|
||
|
"""
|
||
|
return self.__service_id
|
||
|
|
||
|
def __repr__(self):
|
||
|
return f"{self.__class__.__name__}({self.address!r}, {self.__service_id!r})"
|
||
|
|
||
|
|
||
|
class PoolClosedEvent(_PoolEvent):
|
||
|
"""Published when a Connection Pool is closed.
|
||
|
|
||
|
:Parameters:
|
||
|
- `address`: The address (host, port) pair of the server this Pool is
|
||
|
attempting to connect to.
|
||
|
|
||
|
.. versionadded:: 3.9
|
||
|
"""
|
||
|
|
||
|
__slots__ = ()
|
||
|
|
||
|
|
||
|
class ConnectionClosedReason:
|
||
|
"""An enum that defines values for `reason` on a
|
||
|
:class:`ConnectionClosedEvent`.
|
||
|
|
||
|
.. versionadded:: 3.9
|
||
|
"""
|
||
|
|
||
|
STALE = "stale"
|
||
|
"""The pool was cleared, making the connection no longer valid."""
|
||
|
|
||
|
IDLE = "idle"
|
||
|
"""The connection became stale by being idle for too long (maxIdleTimeMS).
|
||
|
"""
|
||
|
|
||
|
ERROR = "error"
|
||
|
"""The connection experienced an error, making it no longer valid."""
|
||
|
|
||
|
POOL_CLOSED = "poolClosed"
|
||
|
"""The pool was closed, making the connection no longer valid."""
|
||
|
|
||
|
|
||
|
class ConnectionCheckOutFailedReason:
|
||
|
"""An enum that defines values for `reason` on a
|
||
|
:class:`ConnectionCheckOutFailedEvent`.
|
||
|
|
||
|
.. versionadded:: 3.9
|
||
|
"""
|
||
|
|
||
|
TIMEOUT = "timeout"
|
||
|
"""The connection check out attempt exceeded the specified timeout."""
|
||
|
|
||
|
POOL_CLOSED = "poolClosed"
|
||
|
"""The pool was previously closed, and cannot provide new connections."""
|
||
|
|
||
|
CONN_ERROR = "connectionError"
|
||
|
"""The connection check out attempt experienced an error while setting up
|
||
|
a new connection.
|
||
|
"""
|
||
|
|
||
|
|
||
|
class _ConnectionEvent:
|
||
|
"""Private base class for connection events."""
|
||
|
|
||
|
__slots__ = ("__address",)
|
||
|
|
||
|
def __init__(self, address: _Address) -> None:
|
||
|
self.__address = address
|
||
|
|
||
|
@property
|
||
|
def address(self) -> _Address:
|
||
|
"""The address (host, port) pair of the server this connection is
|
||
|
attempting to connect to.
|
||
|
"""
|
||
|
return self.__address
|
||
|
|
||
|
def __repr__(self):
|
||
|
return f"{self.__class__.__name__}({self.__address!r})"
|
||
|
|
||
|
|
||
|
class _ConnectionIdEvent(_ConnectionEvent):
|
||
|
"""Private base class for connection events with an id."""
|
||
|
|
||
|
__slots__ = ("__connection_id",)
|
||
|
|
||
|
def __init__(self, address: _Address, connection_id: int) -> None:
|
||
|
super().__init__(address)
|
||
|
self.__connection_id = connection_id
|
||
|
|
||
|
@property
|
||
|
def connection_id(self) -> int:
|
||
|
"""The ID of the Connection."""
|
||
|
return self.__connection_id
|
||
|
|
||
|
def __repr__(self):
|
||
|
return f"{self.__class__.__name__}({self.address!r}, {self.__connection_id!r})"
|
||
|
|
||
|
|
||
|
class ConnectionCreatedEvent(_ConnectionIdEvent):
|
||
|
"""Published when a Connection Pool creates a Connection object.
|
||
|
|
||
|
NOTE: This connection is not ready for use until the
|
||
|
:class:`ConnectionReadyEvent` is published.
|
||
|
|
||
|
:Parameters:
|
||
|
- `address`: The address (host, port) pair of the server this
|
||
|
Connection is attempting to connect to.
|
||
|
- `connection_id`: The integer ID of the Connection in this Pool.
|
||
|
|
||
|
.. versionadded:: 3.9
|
||
|
"""
|
||
|
|
||
|
__slots__ = ()
|
||
|
|
||
|
|
||
|
class ConnectionReadyEvent(_ConnectionIdEvent):
|
||
|
"""Published when a Connection has finished its setup, and is ready to use.
|
||
|
|
||
|
:Parameters:
|
||
|
- `address`: The address (host, port) pair of the server this
|
||
|
Connection is attempting to connect to.
|
||
|
- `connection_id`: The integer ID of the Connection in this Pool.
|
||
|
|
||
|
.. versionadded:: 3.9
|
||
|
"""
|
||
|
|
||
|
__slots__ = ()
|
||
|
|
||
|
|
||
|
class ConnectionClosedEvent(_ConnectionIdEvent):
|
||
|
"""Published when a Connection is closed.
|
||
|
|
||
|
:Parameters:
|
||
|
- `address`: The address (host, port) pair of the server this
|
||
|
Connection is attempting to connect to.
|
||
|
- `connection_id`: The integer ID of the Connection in this Pool.
|
||
|
- `reason`: A reason explaining why this connection was closed.
|
||
|
|
||
|
.. versionadded:: 3.9
|
||
|
"""
|
||
|
|
||
|
__slots__ = ("__reason",)
|
||
|
|
||
|
def __init__(self, address, connection_id, reason):
|
||
|
super().__init__(address, connection_id)
|
||
|
self.__reason = reason
|
||
|
|
||
|
@property
|
||
|
def reason(self):
|
||
|
"""A reason explaining why this connection was closed.
|
||
|
|
||
|
The reason must be one of the strings from the
|
||
|
:class:`ConnectionClosedReason` enum.
|
||
|
"""
|
||
|
return self.__reason
|
||
|
|
||
|
def __repr__(self):
|
||
|
return "{}({!r}, {!r}, {!r})".format(
|
||
|
self.__class__.__name__,
|
||
|
self.address,
|
||
|
self.connection_id,
|
||
|
self.__reason,
|
||
|
)
|
||
|
|
||
|
|
||
|
class ConnectionCheckOutStartedEvent(_ConnectionEvent):
|
||
|
"""Published when the driver starts attempting to check out a connection.
|
||
|
|
||
|
:Parameters:
|
||
|
- `address`: The address (host, port) pair of the server this
|
||
|
Connection is attempting to connect to.
|
||
|
|
||
|
.. versionadded:: 3.9
|
||
|
"""
|
||
|
|
||
|
__slots__ = ()
|
||
|
|
||
|
|
||
|
class ConnectionCheckOutFailedEvent(_ConnectionEvent):
|
||
|
"""Published when the driver's attempt to check out a connection fails.
|
||
|
|
||
|
:Parameters:
|
||
|
- `address`: The address (host, port) pair of the server this
|
||
|
Connection is attempting to connect to.
|
||
|
- `reason`: A reason explaining why connection check out failed.
|
||
|
|
||
|
.. versionadded:: 3.9
|
||
|
"""
|
||
|
|
||
|
__slots__ = ("__reason",)
|
||
|
|
||
|
def __init__(self, address: _Address, reason: str) -> None:
|
||
|
super().__init__(address)
|
||
|
self.__reason = reason
|
||
|
|
||
|
@property
|
||
|
def reason(self) -> str:
|
||
|
"""A reason explaining why connection check out failed.
|
||
|
|
||
|
The reason must be one of the strings from the
|
||
|
:class:`ConnectionCheckOutFailedReason` enum.
|
||
|
"""
|
||
|
return self.__reason
|
||
|
|
||
|
def __repr__(self):
|
||
|
return f"{self.__class__.__name__}({self.address!r}, {self.__reason!r})"
|
||
|
|
||
|
|
||
|
class ConnectionCheckedOutEvent(_ConnectionIdEvent):
|
||
|
"""Published when the driver successfully checks out a Connection.
|
||
|
|
||
|
:Parameters:
|
||
|
- `address`: The address (host, port) pair of the server this
|
||
|
Connection is attempting to connect to.
|
||
|
- `connection_id`: The integer ID of the Connection in this Pool.
|
||
|
|
||
|
.. versionadded:: 3.9
|
||
|
"""
|
||
|
|
||
|
__slots__ = ()
|
||
|
|
||
|
|
||
|
class ConnectionCheckedInEvent(_ConnectionIdEvent):
|
||
|
"""Published when the driver checks in a Connection into the Pool.
|
||
|
|
||
|
:Parameters:
|
||
|
- `address`: The address (host, port) pair of the server this
|
||
|
Connection is attempting to connect to.
|
||
|
- `connection_id`: The integer ID of the Connection in this Pool.
|
||
|
|
||
|
.. versionadded:: 3.9
|
||
|
"""
|
||
|
|
||
|
__slots__ = ()
|
||
|
|
||
|
|
||
|
class _ServerEvent:
|
||
|
"""Base class for server events."""
|
||
|
|
||
|
__slots__ = ("__server_address", "__topology_id")
|
||
|
|
||
|
def __init__(self, server_address: _Address, topology_id: ObjectId) -> None:
|
||
|
self.__server_address = server_address
|
||
|
self.__topology_id = topology_id
|
||
|
|
||
|
@property
|
||
|
def server_address(self) -> _Address:
|
||
|
"""The address (host, port) pair of the server"""
|
||
|
return self.__server_address
|
||
|
|
||
|
@property
|
||
|
def topology_id(self) -> ObjectId:
|
||
|
"""A unique identifier for the topology this server is a part of."""
|
||
|
return self.__topology_id
|
||
|
|
||
|
def __repr__(self):
|
||
|
return "<{} {} topology_id: {}>".format(
|
||
|
self.__class__.__name__,
|
||
|
self.server_address,
|
||
|
self.topology_id,
|
||
|
)
|
||
|
|
||
|
|
||
|
class ServerDescriptionChangedEvent(_ServerEvent):
|
||
|
"""Published when server description changes.
|
||
|
|
||
|
.. versionadded:: 3.3
|
||
|
"""
|
||
|
|
||
|
__slots__ = ("__previous_description", "__new_description")
|
||
|
|
||
|
def __init__(
|
||
|
self,
|
||
|
previous_description: ServerDescription,
|
||
|
new_description: ServerDescription,
|
||
|
*args: Any,
|
||
|
) -> None:
|
||
|
super().__init__(*args)
|
||
|
self.__previous_description = previous_description
|
||
|
self.__new_description = new_description
|
||
|
|
||
|
@property
|
||
|
def previous_description(self) -> ServerDescription:
|
||
|
"""The previous
|
||
|
:class:`~pymongo.server_description.ServerDescription`.
|
||
|
"""
|
||
|
return self.__previous_description
|
||
|
|
||
|
@property
|
||
|
def new_description(self) -> ServerDescription:
|
||
|
"""The new
|
||
|
:class:`~pymongo.server_description.ServerDescription`.
|
||
|
"""
|
||
|
return self.__new_description
|
||
|
|
||
|
def __repr__(self):
|
||
|
return "<{} {} changed from: {}, to: {}>".format(
|
||
|
self.__class__.__name__,
|
||
|
self.server_address,
|
||
|
self.previous_description,
|
||
|
self.new_description,
|
||
|
)
|
||
|
|
||
|
|
||
|
class ServerOpeningEvent(_ServerEvent):
|
||
|
"""Published when server is initialized.
|
||
|
|
||
|
.. versionadded:: 3.3
|
||
|
"""
|
||
|
|
||
|
__slots__ = ()
|
||
|
|
||
|
|
||
|
class ServerClosedEvent(_ServerEvent):
|
||
|
"""Published when server is closed.
|
||
|
|
||
|
.. versionadded:: 3.3
|
||
|
"""
|
||
|
|
||
|
__slots__ = ()
|
||
|
|
||
|
|
||
|
class TopologyEvent:
|
||
|
"""Base class for topology description events."""
|
||
|
|
||
|
__slots__ = "__topology_id"
|
||
|
|
||
|
def __init__(self, topology_id: ObjectId) -> None:
|
||
|
self.__topology_id = topology_id
|
||
|
|
||
|
@property
|
||
|
def topology_id(self) -> ObjectId:
|
||
|
"""A unique identifier for the topology this server is a part of."""
|
||
|
return self.__topology_id
|
||
|
|
||
|
def __repr__(self):
|
||
|
return f"<{self.__class__.__name__} topology_id: {self.topology_id}>"
|
||
|
|
||
|
|
||
|
class TopologyDescriptionChangedEvent(TopologyEvent):
|
||
|
"""Published when the topology description changes.
|
||
|
|
||
|
.. versionadded:: 3.3
|
||
|
"""
|
||
|
|
||
|
__slots__ = ("__previous_description", "__new_description")
|
||
|
|
||
|
def __init__(
|
||
|
self,
|
||
|
previous_description: TopologyDescription,
|
||
|
new_description: TopologyDescription,
|
||
|
*args: Any,
|
||
|
) -> None:
|
||
|
super().__init__(*args)
|
||
|
self.__previous_description = previous_description
|
||
|
self.__new_description = new_description
|
||
|
|
||
|
@property
|
||
|
def previous_description(self) -> TopologyDescription:
|
||
|
"""The previous
|
||
|
:class:`~pymongo.topology_description.TopologyDescription`.
|
||
|
"""
|
||
|
return self.__previous_description
|
||
|
|
||
|
@property
|
||
|
def new_description(self) -> TopologyDescription:
|
||
|
"""The new
|
||
|
:class:`~pymongo.topology_description.TopologyDescription`.
|
||
|
"""
|
||
|
return self.__new_description
|
||
|
|
||
|
def __repr__(self):
|
||
|
return "<{} topology_id: {} changed from: {}, to: {}>".format(
|
||
|
self.__class__.__name__,
|
||
|
self.topology_id,
|
||
|
self.previous_description,
|
||
|
self.new_description,
|
||
|
)
|
||
|
|
||
|
|
||
|
class TopologyOpenedEvent(TopologyEvent):
|
||
|
"""Published when the topology is initialized.
|
||
|
|
||
|
.. versionadded:: 3.3
|
||
|
"""
|
||
|
|
||
|
__slots__ = ()
|
||
|
|
||
|
|
||
|
class TopologyClosedEvent(TopologyEvent):
|
||
|
"""Published when the topology is closed.
|
||
|
|
||
|
.. versionadded:: 3.3
|
||
|
"""
|
||
|
|
||
|
__slots__ = ()
|
||
|
|
||
|
|
||
|
class _ServerHeartbeatEvent:
|
||
|
"""Base class for server heartbeat events."""
|
||
|
|
||
|
__slots__ = "__connection_id"
|
||
|
|
||
|
def __init__(self, connection_id: _Address) -> None:
|
||
|
self.__connection_id = connection_id
|
||
|
|
||
|
@property
|
||
|
def connection_id(self) -> _Address:
|
||
|
"""The address (host, port) of the server this heartbeat was sent
|
||
|
to.
|
||
|
"""
|
||
|
return self.__connection_id
|
||
|
|
||
|
def __repr__(self):
|
||
|
return f"<{self.__class__.__name__} {self.connection_id}>"
|
||
|
|
||
|
|
||
|
class ServerHeartbeatStartedEvent(_ServerHeartbeatEvent):
|
||
|
"""Published when a heartbeat is started.
|
||
|
|
||
|
.. versionadded:: 3.3
|
||
|
"""
|
||
|
|
||
|
__slots__ = ()
|
||
|
|
||
|
|
||
|
class ServerHeartbeatSucceededEvent(_ServerHeartbeatEvent):
|
||
|
"""Fired when the server heartbeat succeeds.
|
||
|
|
||
|
.. versionadded:: 3.3
|
||
|
"""
|
||
|
|
||
|
__slots__ = ("__duration", "__reply", "__awaited")
|
||
|
|
||
|
def __init__(
|
||
|
self, duration: float, reply: Hello, connection_id: _Address, awaited: bool = False
|
||
|
) -> None:
|
||
|
super().__init__(connection_id)
|
||
|
self.__duration = duration
|
||
|
self.__reply = reply
|
||
|
self.__awaited = awaited
|
||
|
|
||
|
@property
|
||
|
def duration(self) -> float:
|
||
|
"""The duration of this heartbeat in microseconds."""
|
||
|
return self.__duration
|
||
|
|
||
|
@property
|
||
|
def reply(self) -> Hello:
|
||
|
"""An instance of :class:`~pymongo.hello.Hello`."""
|
||
|
return self.__reply
|
||
|
|
||
|
@property
|
||
|
def awaited(self) -> bool:
|
||
|
"""Whether the heartbeat was awaited.
|
||
|
|
||
|
If true, then :meth:`duration` reflects the sum of the round trip time
|
||
|
to the server and the time that the server waited before sending a
|
||
|
response.
|
||
|
"""
|
||
|
return self.__awaited
|
||
|
|
||
|
def __repr__(self):
|
||
|
return "<{} {} duration: {}, awaited: {}, reply: {}>".format(
|
||
|
self.__class__.__name__,
|
||
|
self.connection_id,
|
||
|
self.duration,
|
||
|
self.awaited,
|
||
|
self.reply,
|
||
|
)
|
||
|
|
||
|
|
||
|
class ServerHeartbeatFailedEvent(_ServerHeartbeatEvent):
|
||
|
"""Fired when the server heartbeat fails, either with an "ok: 0"
|
||
|
or a socket exception.
|
||
|
|
||
|
.. versionadded:: 3.3
|
||
|
"""
|
||
|
|
||
|
__slots__ = ("__duration", "__reply", "__awaited")
|
||
|
|
||
|
def __init__(
|
||
|
self, duration: float, reply: Exception, connection_id: _Address, awaited: bool = False
|
||
|
) -> None:
|
||
|
super().__init__(connection_id)
|
||
|
self.__duration = duration
|
||
|
self.__reply = reply
|
||
|
self.__awaited = awaited
|
||
|
|
||
|
@property
|
||
|
def duration(self) -> float:
|
||
|
"""The duration of this heartbeat in microseconds."""
|
||
|
return self.__duration
|
||
|
|
||
|
@property
|
||
|
def reply(self) -> Exception:
|
||
|
"""A subclass of :exc:`Exception`."""
|
||
|
return self.__reply
|
||
|
|
||
|
@property
|
||
|
def awaited(self) -> bool:
|
||
|
"""Whether the heartbeat was awaited.
|
||
|
|
||
|
If true, then :meth:`duration` reflects the sum of the round trip time
|
||
|
to the server and the time that the server waited before sending a
|
||
|
response.
|
||
|
"""
|
||
|
return self.__awaited
|
||
|
|
||
|
def __repr__(self):
|
||
|
return "<{} {} duration: {}, awaited: {}, reply: {!r}>".format(
|
||
|
self.__class__.__name__,
|
||
|
self.connection_id,
|
||
|
self.duration,
|
||
|
self.awaited,
|
||
|
self.reply,
|
||
|
)
|
||
|
|
||
|
|
||
|
class _EventListeners:
|
||
|
"""Configure event listeners for a client instance.
|
||
|
|
||
|
Any event listeners registered globally are included by default.
|
||
|
|
||
|
:Parameters:
|
||
|
- `listeners`: A list of event listeners.
|
||
|
"""
|
||
|
|
||
|
def __init__(self, listeners):
|
||
|
self.__command_listeners = _LISTENERS.command_listeners[:]
|
||
|
self.__server_listeners = _LISTENERS.server_listeners[:]
|
||
|
lst = _LISTENERS.server_heartbeat_listeners
|
||
|
self.__server_heartbeat_listeners = lst[:]
|
||
|
self.__topology_listeners = _LISTENERS.topology_listeners[:]
|
||
|
self.__cmap_listeners = _LISTENERS.cmap_listeners[:]
|
||
|
if listeners is not None:
|
||
|
for lst in listeners:
|
||
|
if isinstance(lst, CommandListener):
|
||
|
self.__command_listeners.append(lst)
|
||
|
if isinstance(lst, ServerListener):
|
||
|
self.__server_listeners.append(lst)
|
||
|
if isinstance(lst, ServerHeartbeatListener):
|
||
|
self.__server_heartbeat_listeners.append(lst)
|
||
|
if isinstance(lst, TopologyListener):
|
||
|
self.__topology_listeners.append(lst)
|
||
|
if isinstance(lst, ConnectionPoolListener):
|
||
|
self.__cmap_listeners.append(lst)
|
||
|
self.__enabled_for_commands = bool(self.__command_listeners)
|
||
|
self.__enabled_for_server = bool(self.__server_listeners)
|
||
|
self.__enabled_for_server_heartbeat = bool(self.__server_heartbeat_listeners)
|
||
|
self.__enabled_for_topology = bool(self.__topology_listeners)
|
||
|
self.__enabled_for_cmap = bool(self.__cmap_listeners)
|
||
|
|
||
|
@property
|
||
|
def enabled_for_commands(self):
|
||
|
"""Are any CommandListener instances registered?"""
|
||
|
return self.__enabled_for_commands
|
||
|
|
||
|
@property
|
||
|
def enabled_for_server(self):
|
||
|
"""Are any ServerListener instances registered?"""
|
||
|
return self.__enabled_for_server
|
||
|
|
||
|
@property
|
||
|
def enabled_for_server_heartbeat(self):
|
||
|
"""Are any ServerHeartbeatListener instances registered?"""
|
||
|
return self.__enabled_for_server_heartbeat
|
||
|
|
||
|
@property
|
||
|
def enabled_for_topology(self):
|
||
|
"""Are any TopologyListener instances registered?"""
|
||
|
return self.__enabled_for_topology
|
||
|
|
||
|
@property
|
||
|
def enabled_for_cmap(self):
|
||
|
"""Are any ConnectionPoolListener instances registered?"""
|
||
|
return self.__enabled_for_cmap
|
||
|
|
||
|
def event_listeners(self):
|
||
|
"""List of registered event listeners."""
|
||
|
return (
|
||
|
self.__command_listeners
|
||
|
+ self.__server_heartbeat_listeners
|
||
|
+ self.__server_listeners
|
||
|
+ self.__topology_listeners
|
||
|
+ self.__cmap_listeners
|
||
|
)
|
||
|
|
||
|
def publish_command_start(
|
||
|
self, command, database_name, request_id, connection_id, op_id=None, service_id=None
|
||
|
):
|
||
|
"""Publish a CommandStartedEvent to all command listeners.
|
||
|
|
||
|
:Parameters:
|
||
|
- `command`: The command document.
|
||
|
- `database_name`: The name of the database this command was run
|
||
|
against.
|
||
|
- `request_id`: The request id for this operation.
|
||
|
- `connection_id`: The address (host, port) of the server this
|
||
|
command was sent to.
|
||
|
- `op_id`: The (optional) operation id for this operation.
|
||
|
- `service_id`: The service_id this command was sent to, or ``None``.
|
||
|
"""
|
||
|
if op_id is None:
|
||
|
op_id = request_id
|
||
|
event = CommandStartedEvent(
|
||
|
command, database_name, request_id, connection_id, op_id, service_id=service_id
|
||
|
)
|
||
|
for subscriber in self.__command_listeners:
|
||
|
try:
|
||
|
subscriber.started(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_command_success(
|
||
|
self,
|
||
|
duration,
|
||
|
reply,
|
||
|
command_name,
|
||
|
request_id,
|
||
|
connection_id,
|
||
|
op_id=None,
|
||
|
service_id=None,
|
||
|
speculative_hello=False,
|
||
|
):
|
||
|
"""Publish a CommandSucceededEvent to all command listeners.
|
||
|
|
||
|
:Parameters:
|
||
|
- `duration`: The command duration as a datetime.timedelta.
|
||
|
- `reply`: The server reply document.
|
||
|
- `command_name`: The command name.
|
||
|
- `request_id`: The request id for this operation.
|
||
|
- `connection_id`: The address (host, port) of the server this
|
||
|
command was sent to.
|
||
|
- `op_id`: The (optional) operation id for this operation.
|
||
|
- `service_id`: The service_id this command was sent to, or ``None``.
|
||
|
- `speculative_hello`: Was the command sent with speculative auth?
|
||
|
"""
|
||
|
if op_id is None:
|
||
|
op_id = request_id
|
||
|
if speculative_hello:
|
||
|
# Redact entire response when the command started contained
|
||
|
# speculativeAuthenticate.
|
||
|
reply = {}
|
||
|
event = CommandSucceededEvent(
|
||
|
duration, reply, command_name, request_id, connection_id, op_id, service_id
|
||
|
)
|
||
|
for subscriber in self.__command_listeners:
|
||
|
try:
|
||
|
subscriber.succeeded(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_command_failure(
|
||
|
self,
|
||
|
duration,
|
||
|
failure,
|
||
|
command_name,
|
||
|
request_id,
|
||
|
connection_id,
|
||
|
op_id=None,
|
||
|
service_id=None,
|
||
|
):
|
||
|
"""Publish a CommandFailedEvent to all command listeners.
|
||
|
|
||
|
:Parameters:
|
||
|
- `duration`: The command duration as a datetime.timedelta.
|
||
|
- `failure`: The server reply document or failure description
|
||
|
document.
|
||
|
- `command_name`: The command name.
|
||
|
- `request_id`: The request id for this operation.
|
||
|
- `connection_id`: The address (host, port) of the server this
|
||
|
command was sent to.
|
||
|
- `op_id`: The (optional) operation id for this operation.
|
||
|
- `service_id`: The service_id this command was sent to, or ``None``.
|
||
|
"""
|
||
|
if op_id is None:
|
||
|
op_id = request_id
|
||
|
event = CommandFailedEvent(
|
||
|
duration, failure, command_name, request_id, connection_id, op_id, service_id=service_id
|
||
|
)
|
||
|
for subscriber in self.__command_listeners:
|
||
|
try:
|
||
|
subscriber.failed(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_server_heartbeat_started(self, connection_id):
|
||
|
"""Publish a ServerHeartbeatStartedEvent to all server heartbeat
|
||
|
listeners.
|
||
|
|
||
|
:Parameters:
|
||
|
- `connection_id`: The address (host, port) pair of the connection.
|
||
|
"""
|
||
|
event = ServerHeartbeatStartedEvent(connection_id)
|
||
|
for subscriber in self.__server_heartbeat_listeners:
|
||
|
try:
|
||
|
subscriber.started(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_server_heartbeat_succeeded(self, connection_id, duration, reply, awaited):
|
||
|
"""Publish a ServerHeartbeatSucceededEvent to all server heartbeat
|
||
|
listeners.
|
||
|
|
||
|
:Parameters:
|
||
|
- `connection_id`: The address (host, port) pair of the connection.
|
||
|
- `duration`: The execution time of the event in the highest possible
|
||
|
resolution for the platform.
|
||
|
- `reply`: The command reply.
|
||
|
- `awaited`: True if the response was awaited.
|
||
|
"""
|
||
|
event = ServerHeartbeatSucceededEvent(duration, reply, connection_id, awaited)
|
||
|
for subscriber in self.__server_heartbeat_listeners:
|
||
|
try:
|
||
|
subscriber.succeeded(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_server_heartbeat_failed(self, connection_id, duration, reply, awaited):
|
||
|
"""Publish a ServerHeartbeatFailedEvent to all server heartbeat
|
||
|
listeners.
|
||
|
|
||
|
:Parameters:
|
||
|
- `connection_id`: The address (host, port) pair of the connection.
|
||
|
- `duration`: The execution time of the event in the highest possible
|
||
|
resolution for the platform.
|
||
|
- `reply`: The command reply.
|
||
|
- `awaited`: True if the response was awaited.
|
||
|
"""
|
||
|
event = ServerHeartbeatFailedEvent(duration, reply, connection_id, awaited)
|
||
|
for subscriber in self.__server_heartbeat_listeners:
|
||
|
try:
|
||
|
subscriber.failed(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_server_opened(self, server_address, topology_id):
|
||
|
"""Publish a ServerOpeningEvent to all server listeners.
|
||
|
|
||
|
:Parameters:
|
||
|
- `server_address`: The address (host, port) pair of the server.
|
||
|
- `topology_id`: A unique identifier for the topology this server
|
||
|
is a part of.
|
||
|
"""
|
||
|
event = ServerOpeningEvent(server_address, topology_id)
|
||
|
for subscriber in self.__server_listeners:
|
||
|
try:
|
||
|
subscriber.opened(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_server_closed(self, server_address, topology_id):
|
||
|
"""Publish a ServerClosedEvent to all server listeners.
|
||
|
|
||
|
:Parameters:
|
||
|
- `server_address`: The address (host, port) pair of the server.
|
||
|
- `topology_id`: A unique identifier for the topology this server
|
||
|
is a part of.
|
||
|
"""
|
||
|
event = ServerClosedEvent(server_address, topology_id)
|
||
|
for subscriber in self.__server_listeners:
|
||
|
try:
|
||
|
subscriber.closed(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_server_description_changed(
|
||
|
self, previous_description, new_description, server_address, topology_id
|
||
|
):
|
||
|
"""Publish a ServerDescriptionChangedEvent to all server listeners.
|
||
|
|
||
|
:Parameters:
|
||
|
- `previous_description`: The previous server description.
|
||
|
- `server_address`: The address (host, port) pair of the server.
|
||
|
- `new_description`: The new server description.
|
||
|
- `topology_id`: A unique identifier for the topology this server
|
||
|
is a part of.
|
||
|
"""
|
||
|
event = ServerDescriptionChangedEvent(
|
||
|
previous_description, new_description, server_address, topology_id
|
||
|
)
|
||
|
for subscriber in self.__server_listeners:
|
||
|
try:
|
||
|
subscriber.description_changed(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_topology_opened(self, topology_id):
|
||
|
"""Publish a TopologyOpenedEvent to all topology listeners.
|
||
|
|
||
|
:Parameters:
|
||
|
- `topology_id`: A unique identifier for the topology this server
|
||
|
is a part of.
|
||
|
"""
|
||
|
event = TopologyOpenedEvent(topology_id)
|
||
|
for subscriber in self.__topology_listeners:
|
||
|
try:
|
||
|
subscriber.opened(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_topology_closed(self, topology_id):
|
||
|
"""Publish a TopologyClosedEvent to all topology listeners.
|
||
|
|
||
|
:Parameters:
|
||
|
- `topology_id`: A unique identifier for the topology this server
|
||
|
is a part of.
|
||
|
"""
|
||
|
event = TopologyClosedEvent(topology_id)
|
||
|
for subscriber in self.__topology_listeners:
|
||
|
try:
|
||
|
subscriber.closed(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_topology_description_changed(
|
||
|
self, previous_description, new_description, topology_id
|
||
|
):
|
||
|
"""Publish a TopologyDescriptionChangedEvent to all topology listeners.
|
||
|
|
||
|
:Parameters:
|
||
|
- `previous_description`: The previous topology description.
|
||
|
- `new_description`: The new topology description.
|
||
|
- `topology_id`: A unique identifier for the topology this server
|
||
|
is a part of.
|
||
|
"""
|
||
|
event = TopologyDescriptionChangedEvent(previous_description, new_description, topology_id)
|
||
|
for subscriber in self.__topology_listeners:
|
||
|
try:
|
||
|
subscriber.description_changed(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_pool_created(self, address, options):
|
||
|
"""Publish a :class:`PoolCreatedEvent` to all pool listeners."""
|
||
|
event = PoolCreatedEvent(address, options)
|
||
|
for subscriber in self.__cmap_listeners:
|
||
|
try:
|
||
|
subscriber.pool_created(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_pool_ready(self, address):
|
||
|
"""Publish a :class:`PoolReadyEvent` to all pool listeners."""
|
||
|
event = PoolReadyEvent(address)
|
||
|
for subscriber in self.__cmap_listeners:
|
||
|
try:
|
||
|
subscriber.pool_ready(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_pool_cleared(self, address, service_id):
|
||
|
"""Publish a :class:`PoolClearedEvent` to all pool listeners."""
|
||
|
event = PoolClearedEvent(address, service_id)
|
||
|
for subscriber in self.__cmap_listeners:
|
||
|
try:
|
||
|
subscriber.pool_cleared(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_pool_closed(self, address):
|
||
|
"""Publish a :class:`PoolClosedEvent` to all pool listeners."""
|
||
|
event = PoolClosedEvent(address)
|
||
|
for subscriber in self.__cmap_listeners:
|
||
|
try:
|
||
|
subscriber.pool_closed(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_connection_created(self, address, connection_id):
|
||
|
"""Publish a :class:`ConnectionCreatedEvent` to all connection
|
||
|
listeners.
|
||
|
"""
|
||
|
event = ConnectionCreatedEvent(address, connection_id)
|
||
|
for subscriber in self.__cmap_listeners:
|
||
|
try:
|
||
|
subscriber.connection_created(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_connection_ready(self, address, connection_id):
|
||
|
"""Publish a :class:`ConnectionReadyEvent` to all connection listeners."""
|
||
|
event = ConnectionReadyEvent(address, connection_id)
|
||
|
for subscriber in self.__cmap_listeners:
|
||
|
try:
|
||
|
subscriber.connection_ready(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_connection_closed(self, address, connection_id, reason):
|
||
|
"""Publish a :class:`ConnectionClosedEvent` to all connection
|
||
|
listeners.
|
||
|
"""
|
||
|
event = ConnectionClosedEvent(address, connection_id, reason)
|
||
|
for subscriber in self.__cmap_listeners:
|
||
|
try:
|
||
|
subscriber.connection_closed(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_connection_check_out_started(self, address):
|
||
|
"""Publish a :class:`ConnectionCheckOutStartedEvent` to all connection
|
||
|
listeners.
|
||
|
"""
|
||
|
event = ConnectionCheckOutStartedEvent(address)
|
||
|
for subscriber in self.__cmap_listeners:
|
||
|
try:
|
||
|
subscriber.connection_check_out_started(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_connection_check_out_failed(self, address, reason):
|
||
|
"""Publish a :class:`ConnectionCheckOutFailedEvent` to all connection
|
||
|
listeners.
|
||
|
"""
|
||
|
event = ConnectionCheckOutFailedEvent(address, reason)
|
||
|
for subscriber in self.__cmap_listeners:
|
||
|
try:
|
||
|
subscriber.connection_check_out_failed(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_connection_checked_out(self, address, connection_id):
|
||
|
"""Publish a :class:`ConnectionCheckedOutEvent` to all connection
|
||
|
listeners.
|
||
|
"""
|
||
|
event = ConnectionCheckedOutEvent(address, connection_id)
|
||
|
for subscriber in self.__cmap_listeners:
|
||
|
try:
|
||
|
subscriber.connection_checked_out(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|
||
|
|
||
|
def publish_connection_checked_in(self, address, connection_id):
|
||
|
"""Publish a :class:`ConnectionCheckedInEvent` to all connection
|
||
|
listeners.
|
||
|
"""
|
||
|
event = ConnectionCheckedInEvent(address, connection_id)
|
||
|
for subscriber in self.__cmap_listeners:
|
||
|
try:
|
||
|
subscriber.connection_checked_in(event)
|
||
|
except Exception:
|
||
|
_handle_exception()
|