pittgoogle.schema

Classes to manage alert schemas.

`Serializers`()
`Schema`(*, name, description, origin, ...[, ...])	Class for an individual schema.
`DefaultSchema`(*, name, description, origin)	Default schema to serialize and deserialize alert bytes.
`ElasticcSchema`(*, name, description, origin)	Schema for ELAsTiCC alerts.
`LsstSchema`(*, name, description, origin[, ...])	Schema for LSST alerts.
`LvkSchema`(*, name, description, origin[, ...])	Schema for LVK alerts.
`ZtfSchema`(*, name, description, origin[, ...])	Schema for ZTF alerts.

class pittgoogle.schema.DefaultSchema(*, name: str, description: str, origin: str, version: str | None = None, version_id: str | None = None, definition: dict | None = None, path: Path | None = None, filter_map: dict = NOTHING, serializer: Literal['json', 'avro'] = 'json', deserializer: Literal['json', 'avro'] = 'json')[source]

Default schema to serialize and deserialize alert bytes.

deserialize(alert_bytes: bytes) → dict[source]

Deserialize alert_bytes using JSON or Avro format as defined by the deserializer property.

Parameters:: alert_bytes (bytes) – The bytes to be deserialized.
Returns:: A dictionary representing the deserialized alert_bytes.

deserializer: Literal['json', 'avro']: Whether to use a JSON (default) or Avro to deserialize when decoding alert_bytes -> alert_dict. If “avro”, this pittgoogle.Schema will expect the Avro schema to be attached to alert_bytes in the header.

serialize(alert_dict: dict, *, serializer: Literal['json', 'avro', None] = None) → bytes[source]

Serialize the alert_dict.

Parameters:

alert_dict (dict) – The dictionary to be serialized.
serializer (str or None, optional) – Whether to serialize the dict using Avro or JSON. If not None, this will override the serializer property and is subject to the same conditions.

Returns:

The serialized data in bytes.

Return type:

bytes

serializer: Literal['json', 'avro']: Whether to serialize the alert_dict to JSON (default) or Avro when, e.g., publishing a Pub/Sub message. If “avro”, the user must supply the schema definition by setting Schema.definition().

class pittgoogle.schema.ElasticcSchema(*, name: str, description: str, origin: str, version: str | None = None, version_id: str | None = None, definition: dict | None = None, path: Path | None = None, filter_map: dict = NOTHING, serializer: Literal['json', 'avro'] = 'avro', deserializer: Literal['json', 'avro'] = 'avro')[source]

Schema for ELAsTiCC alerts.

deserialize(alert_bytes: bytes) → dict[source]

Deserialize alert_bytes using JSON or Avro format as defined by ElasticcSchema.deserializer().

Parameters:: alert_bytes (bytes) – The bytes to be deserialized.
Returns:: A dictionary representing the deserialized alert_bytes.

deserializer: Literal['json', 'avro']: Whether to use a Avro (default) or JSON to deserialize when decoding alert_bytes -> alert_dict. If “avro”, this schema will use the schemaless Avro format.

serialize(alert_dict: dict, *, serializer: Literal['json', 'avro', None] = None) → bytes[source]

Serialize the alert_dict.

Parameters:

alert_dict (dict) – The dictionary to be serialized.
serializer (str or None, optional) – Whether to serialize the dict using Avro or JSON. If not None, this will override ElasticcSchema.serializer() and is subject to the same conditions.

Returns:

The serialized data in bytes.

Return type:

bytes

serializer: Literal['json', 'avro']: Whether to serialize the dict to Avro (default) or JSON when, e.g., publishing a Pub/Sub message. If “avro”, this schema will use the schemaless Avro format.

class pittgoogle.schema.LsstSchema(*, name: str, description: str, origin: str, version: str | None = None, version_id: str | None = None, definition: dict | None = None, path: Path | None = None, filter_map: dict = NOTHING, serializer: Literal['json', 'avro'] = 'avro', deserializer: Literal['json', 'avro'] = 'avro')[source]

Schema for LSST alerts.

deserialize(alert_bytes: bytes) → dict[source]

Deserialize alert_bytes using JSON or Avro format as defined by LsstSchema.deserializer().

Parameters:: alert_bytes (bytes) – The bytes to be deserialized.
Returns:: A dictionary representing the deserialized alert_bytes.

deserializer: Literal['json', 'avro']: Whether to use Avro (default) or JSON to deserialize when decoding alert_bytes -> alert_dict. If “avro”, this schema will use the Avro Confluent Wire Format (https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format).

serialize(alert_dict: dict, *, serializer: Literal['json', 'avro', None] = None) → bytes[source]

Serialize the alert_dict.

Parameters:

alert_dict (dict) – The dictionary to be serialized.
serializer (str or None, optional) – Whether to serialize the dict using Avro or JSON. If not None, this will override LsstSchema.serializer() and is subject to the same conditions.

Returns:

The serialized data in bytes.

Return type:

bytes

Raises:

exceptions.SchemaError – If the schema version or definition are unavailable and Avro serialization is requested.

serializer: Literal['json', 'avro']: Whether to serialize the dict to Avro (default) or JSON when, e.g., publishing a Pub/Sub message. If “avro”, this schema will use the Avro Confluent Wire Format (https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format).

class pittgoogle.schema.LvkSchema(*, name: str, description: str, origin: str, version: str | None = None, version_id: str | None = None, definition: dict | None = None, path: Path | None = None, filter_map: dict = NOTHING, serializer: Literal['json', 'avro'] = 'json', deserializer: Literal['json', 'avro'] = 'json')[source]: Schema for LVK alerts.

class pittgoogle.schema.Schema(*, name: str, description: str, origin: str, serializer: Literal['json', 'avro'], deserializer: Literal['json', 'avro'], version: str | None = None, version_id: str | None = None, definition: dict | None = None, path: Path | None = None, filter_map: dict = NOTHING)[source]

Class for an individual schema.

Do not call this class’s constructor directly. Instead, load a schema using the registry pittgoogle.registry.Schemas.

definition: dict | None: The schema definition used to serialize and deserialize the alert bytes, if one is required.

description: str: A description of the schema.

abstractmethod deserialize(alert_bytes: bytes) → dict[source]

Deserialize alert_bytes. This method must be implemented by subclasses.

Parameters:: alert_bytes (bytes) – The bytes to be deserialized.
Returns:: A dictionary representing the deserialized alert_bytes.
Return type:: dict

deserializer: Literal['json', 'avro']: Whether to use a JSON or Avro to deserialize when decoding alert_bytes -> alert_dict.

filter_map: dict: Mapping of the filter name as stored in the alert (often an int) to the common name (often a string).

property map: dict: Mapping of Pitt-Google’s generic field names to survey-specific field names.

name: str: Name of the schema. This is typically the name of the survey as well.

origin: str: Pointer to the schema’s origin. Typically this is a URL to a repo maintained by the survey.

path: Path | None: Path to a file containing the schema definition.

abstractmethod serialize(alert_dict: dict, *, serializer: Literal['json', 'avro', None] = None) → bytes[source]

Serialize alert_dict. This method must be implemented by subclasses.

Parameters:

alert_dict (dict) – The dictionary to be serialized.
serializer (str or None, optional) – Whether to serialize the dict using Avro or JSON. If not None, this will override the serializer property and is subject to the same conditions.

Returns:

The serialized data in bytes.

Return type:

bytes

serializer: Literal['json', 'avro']: Whether to serialize the dict to JSON or Avro when, e.g., publishing a Pub/Sub message.

version: str | None: Version of the schema, or None.

version_id: str | None

_ConfluentWireAvroSchema.

Type:: Version ID of the schema, or None. Currently only used for class

class pittgoogle.schema.ZtfSchema(*, name: str, description: str, origin: str, version: str | None = None, version_id: str | None = None, definition: dict | None = None, path: Path | None = None, filter_map: dict = NOTHING, serializer: Literal['json', 'avro'] = 'json', deserializer: Literal['json', 'avro'] = 'avro')[source]

Schema for ZTF alerts.

deserializer: Literal['json', 'avro']: Whether to use a Avro (default) or JSON to deserialize when decoding alert_bytes -> alert_dict. If “avro”, this pittgoogle.Schema will expect the Avro schema to be attached to alert_bytes in the header.