KafkaMetricsConfiguration

pydantic settings safir.metrics.KafkaMetricsConfiguration

A config model you can pass directly to an EventManager constructor.

This may be easier to use if you don’t need to use kafka, the schema manager, or the schema registry, in any other parts of your application.

Parameters:

Show JSON schema
{
   "title": "KafkaMetricsConfiguration",
   "description": "A config model you can pass directly to an EventManager constructor.\n\nThis may be easier to use if you don't need to use kafka, the schema\nmanager, or the schema registry, in any other parts of your application.",
   "type": "object",
   "properties": {
      "metrics_events": {
         "$ref": "#/$defs/MetricsConfiguration"
      },
      "schema_manager": {
         "$ref": "#/$defs/SchemaManagerSettings"
      },
      "kafka": {
         "$ref": "#/$defs/KafkaConnectionSettings"
      }
   },
   "$defs": {
      "KafkaConnectionSettings": {
         "additionalProperties": false,
         "description": "Settings for connecting to Kafka.\n\nThis settings model supports different authentication methods, which each\nhave different sets of required settings. All of these settings can be\nprovided in ``KAFKA_`` prefixed environment variables. Instances of this\nmodel have properties that can be used to construct different types of\nkafka clients:\n\n.. code-block:: python\n\n   from faststream.broker import KafkaBroker\n\n   from safir.kafka import KafkaConnectionSettings\n\n\n   config = KafkaConnectionSettings()\n   kafka_broker = KafkaBroker(**config.faststream_broker_params)\n\nWhen using this model directly, The ``validated`` property enforces at\nruntime that the correct settings were provided for the desired\nauthentication method, and returns models to access those settings in a\ntype-safe way:\n\n.. code-block:: python\n\n   from pathlib import Path\n\n\n   # ValidationError at runtime: ``client_key_path`` is not provided\n   config = KafkaConnectionSettings(\n       bootstrap_servers=\"something:1234\",\n       security_protocol=KafkaSecurityProtocol.SSL,\n       cluster_ca_path=Path(\"/some/cert.crt\"),\n       client_cert_path=Path(\"/some/other/cert.crt\"),\n   )\n\n   config = KafkaConnectionSettings(\n       bootstrap_servers=\"something:1234\",\n       security_protocol=KafkaSecurityProtocol.SSL,\n       cluster_ca_path=Path(\"/some/path/ca.crt\"),\n       client_cert_path=Path(\"/some/path/user.crt\"),\n       client_key_path=Path(\"/some/path/user.key\"),\n   )\n\n   blah = config.validated.sasl_username  # Static type error",
         "properties": {
            "bootstrap_servers": {
               "description": "A comma-separated list of Kafka brokers to connect to. This should be a list of hostnames or IP addresses, each optionally followed by a port number, separated by commas. ",
               "examples": [
                  "kafka-1:9092,kafka-2:9092,kafka-3:9092",
                  "kafka:9092"
               ],
               "title": "Kafka bootstrap servers",
               "type": "string"
            },
            "security_protocol": {
               "$ref": "#/$defs/SecurityProtocol",
               "description": "The authentication and encryption mode for the connection.",
               "title": "Security Protocol"
            },
            "cluster_ca_path": {
               "anyOf": [
                  {
                     "format": "file-path",
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "description": "The path to the PEM-formatted CA certificate file to use for verifying the broker's certificate. This is only needed for SSL and SASL_SSL security protocols, andeven in those cases, only when the broker's certificate is not signed by a CA trusted by the operating system.",
               "examples": [
                  "/some/dir/ca.crt"
               ],
               "title": "Path to CA certificate file"
            },
            "client_cert_path": {
               "anyOf": [
                  {
                     "format": "file-path",
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "description": "The path to the PEM-formated client certificate file to use for authentication. This is only needed if the broker is configured to require SSL client authentication.",
               "examples": [
                  "/some/dir/user.crt"
               ],
               "title": "Path to client certificate file"
            },
            "client_key_path": {
               "anyOf": [
                  {
                     "format": "file-path",
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "description": "The path to the PEM-formatted client key file to use for authentication. This is only needed if for the SSL securityprotocol.",
               "examples": [
                  "/some/dir/user.key"
               ],
               "title": "Path to client key file"
            },
            "sasl_mechanism": {
               "anyOf": [
                  {
                     "$ref": "#/$defs/SaslMechanism"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "description": "The SASL mechanism to use for authentication. This is only needed for the SASL_SSL and SASL_PLAINTEXT securityprotocols.",
               "title": "SASL mechanism"
            },
            "sasl_username": {
               "anyOf": [
                  {
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "description": "The username to use for SASL authentication. This is only needed for the SASL_SSL and SASL_PLAINTEXT securityprotocols.",
               "title": "SASL username"
            },
            "sasl_password": {
               "anyOf": [
                  {
                     "format": "password",
                     "type": "string",
                     "writeOnly": true
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "description": "The password to use for SASL authentication. This is only needed for the SASL_SSL and SASL_PLAINTEXT securityprotocols.",
               "title": "SASL password"
            }
         },
         "required": [
            "bootstrap_servers",
            "security_protocol"
         ],
         "title": "KafkaConnectionSettings",
         "type": "object"
      },
      "MetricsConfiguration": {
         "additionalProperties": false,
         "description": "Configuration for emitting metrics.",
         "properties": {
            "topic_prefix": {
               "default": "lsst.square.metrics.events",
               "description": "You probably should use the default here. It could be useful in development scenarios to change this.",
               "title": "Metrics topic prefix",
               "type": "string"
            },
            "app_name": {
               "description": "The name of the application that is emitting these metrics",
               "title": "Application name",
               "type": "string"
            },
            "disable": {
               "default": false,
               "description": "Set to \"True\" to prevent actually publishing metrics",
               "title": "Disable",
               "type": "boolean"
            }
         },
         "required": [
            "app_name"
         ],
         "title": "MetricsConfiguration",
         "type": "object"
      },
      "SaslMechanism": {
         "description": "Kafka SASL mechanisms.",
         "enum": [
            "PLAIN",
            "SCRAM-SHA-256",
            "SCRAM-SHA-512"
         ],
         "title": "SaslMechanism",
         "type": "string"
      },
      "SchemaManagerSettings": {
         "additionalProperties": false,
         "description": "Settings for constructing a PydanticSchemaManager.",
         "properties": {
            "registry_url": {
               "description": "URL of a a Confluent-compatible schema registry.",
               "format": "uri",
               "minLength": 1,
               "title": "Schema Registry URL",
               "type": "string"
            },
            "suffix": {
               "default": "",
               "description": "A suffix that is added to the schema name (and thus the subject name). The suffix creates alternate subjects in the Schema Registry so schemas registered during testing and staging don't affect the compatibility continuity of a production subject. For production, it's best to not set a suffix.",
               "examples": [
                  "_dev1"
               ],
               "title": "Suffix",
               "type": "string"
            }
         },
         "required": [
            "registry_url"
         ],
         "title": "SchemaManagerSettings",
         "type": "object"
      },
      "SecurityProtocol": {
         "description": "Kafka SASL security protocols.",
         "enum": [
            "SASL_PLAINTEXT",
            "SASL_SSL",
            "PLAINTEXT",
            "SSL"
         ],
         "title": "SecurityProtocol",
         "type": "string"
      }
   },
   "additionalProperties": false
}

Fields:
field kafka: KafkaConnectionSettings [Optional]
field metrics_events: MetricsConfiguration [Optional]
field schema_manager: SchemaManagerSettings [Optional]
make_manager(logger=None)

Construct an EventManager and all of it’s Kafka dependencies.

If your app doesn’t use Kafka or the Schema Registry, this is a shortcut to getting a working event manager without having to manually construct all of the Kafka dependencies.

Parameters:

logger (BoundLogger | None, default: None) – A logger to use for internal logging

Return type:

EventManager