flow.record.stream

Module Contents

Classes

RecordPrinter	Records are printed as textual representation (repr) to fp.
RecordStreamWriter	Records are written as binary (serialized) to fp.
RecordStreamReader
PathTemplateWriter	Write records to a path on disk, path can be a template string.
RecordArchiver	RecordWriter that writes/archives records to a path with YYYY/mm/dd.
RecordFieldRewriter	Rewrite records using a new RecordDescriptor for chosen fields and/or excluded or new record fields.

Functions

RecordOutput	Return a RecordPrinter if fp is a tty otherwise a RecordStreamWriter.
record_stream	Return a Record stream generator from the given Record sources.

Attributes

log
aRepr

flow.record.stream.log

flow.record.stream.aRepr

flow.record.stream.RecordOutput(fp: IO) → RecordPrinter | RecordStreamWriter

Return a RecordPrinter if fp is a tty otherwise a RecordStreamWriter.

class flow.record.stream.RecordPrinter(fp: BinaryIO, flush: bool = True)

Records are printed as textual representation (repr) to fp.

fp = None

auto_flush = True

write(obj: flow.record.base.Record) → None

flush() → None

close() → None

class flow.record.stream.RecordStreamWriter(fp: BinaryIO)

Records are written as binary (serialized) to fp.

fp = None

packer = None

header_written = False

__del__() → None

on_new_descriptor(descriptor: flow.record.base.RecordDescriptor) → None

close() → None

flush() → None

write(obj: flow.record.base.Record | flow.record.base.RecordDescriptor) → None

writeheader() → None

class flow.record.stream.RecordStreamReader(fp: BinaryIO, selector: str | None = None)

fp = None

recordtype = None

descs = None

packer = None

closed = False

selector = None

readheader() → None

read() → flow.record.base.Record | flow.record.base.RecordDescriptor

close() → None

__iter__() → collections.abc.Iterator[flow.record.base.Record]

flow.record.stream.record_stream(sources: list[str], selector: str | None = None) → collections.abc.Iterator[flow.record.base.Record]

Return a Record stream generator from the given Record sources.

Exceptions in a Record source will be caught so the stream is not interrupted.

class flow.record.stream.PathTemplateWriter(path_template: str | None = None, name: str | None = None)

Write records to a path on disk, path can be a template string.

This allows for archiving records on disk based on timestamp for example.

Default template string is:

‘{name}-{record._generated:%Y%m%dT%H}.records.gz’

Available template fields:

name defaults to “records”, but can be overridden in the initializer. record is the record object ts is record._generated

If the destination path already exists it will rename the existing file using the current datetime.

DEFAULT_TEMPLATE = '{name}-{record._generated:%Y%m%dT%H}.records.gz'

path_template = '{name}-{record._generated:%Y%m%dT%H}.records.gz'

name = 'records'

current_path = None

writer = None

stream = None

rotate_existing_file(path: pathlib.Path) → None

record_stream_for_path(path: str) → flow.record.adapter.AbstractWriter

write(record: flow.record.base.Record) → None

close() → None

class flow.record.stream.RecordArchiver(archive_path: str, path_template: str | None = None, name: str | None = None)

Bases: PathTemplateWriter

RecordWriter that writes/archives records to a path with YYYY/mm/dd.

class flow.record.stream.RecordFieldRewriter(fields: list[str] | None = None, exclude: list[str] | None = None, expression: str | None = None)

Rewrite records using a new RecordDescriptor for chosen fields and/or excluded or new record fields.

fields = []

exclude = []

expression = None

record_descriptor_for_fields

rewrite(record: flow.record.base.Record) → flow.record.base.Record