saf.forward package#

Submodules#

saf.forward.disk module#

The disk forward plugin exists as an implementation example.

It just dumps the collected events to disk

class saf.forward.disk.DiskConfig(*, plugin: str, path: Path, filename: Optional[str] = None, pretty_print: bool = False)[source]#

Bases: ForwardConfigBase

Configuration schema for the disk forward plugin.

path: pathlib.Path#

filename: Optional[str]#

pretty_print: bool#

model_config: ClassVar[ConfigDict] = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'filename': FieldInfo(annotation=Union[str, NoneType], required=False), 'path': FieldInfo(annotation=Path, required=True), 'plugin': FieldInfo(annotation=str, required=True), 'pretty_print': FieldInfo(annotation=bool, required=False, default=False)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

saf.forward.disk.get_config_schema() → Type[DiskConfig][source]#: Get the noop plugin configuration schema.

async saf.forward.disk.forward(*, ctx: PipelineRunContext[DiskConfig], event: CollectedEvent) → None[source]#: Method called to forward the event.

saf.forward.noop module#

The NOOP forward plugin exists as an implementation example.

It doesn’t really do anything to the collected event

saf.forward.noop.get_config_schema() → Type[ForwardConfigBase][source]#: Get the noop plugin configuration schema.

async saf.forward.noop.forward(*, ctx: PipelineRunContext[ForwardConfigBase], event: CollectedEvent) → None[source]#: Method called to forward the event.

saf.forward.test module#

Test forwarder plugin.

The test forward plugin exists as an implementation example and also to be able to test the salt-analytics-framework

It doesn’t really do anything to the collected event.

class saf.forward.test.TestForwardConfig(*, plugin: str, sleep: float = 0.0, path: Optional[Path] = None, message: Optional[str] = None, dump_event: bool = False, add_event_to_shared_cache: bool = False)[source]#

Bases: ForwardConfigBase

Test forwarder configuration.

sleep: float#

path: Optional[pathlib.Path]#

message: Optional[str]#

dump_event: bool#

add_event_to_shared_cache: bool#

model_config: ClassVar[ConfigDict] = {'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'add_event_to_shared_cache': FieldInfo(annotation=bool, required=False, default=False), 'dump_event': FieldInfo(annotation=bool, required=False, default=False), 'message': FieldInfo(annotation=Union[str, NoneType], required=False), 'path': FieldInfo(annotation=Union[Path, NoneType], required=False), 'plugin': FieldInfo(annotation=str, required=True), 'sleep': FieldInfo(annotation=float, required=False, default=0.0)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

saf.forward.test.get_config_schema() → Type[TestForwardConfig][source]#: Get the noop plugin configuration schema.

async saf.forward.test.forward(*, ctx: PipelineRunContext[TestForwardConfig], event: CollectedEvent) → None[source]#: Method called to forward the event.