`flow.record`#

Submodules#

Package Contents#

Classes#

`FieldType`
`GroupedRecord`	GroupedRecord acts like a normal Record, but can contain multiple records.
`Record`
`RecordDescriptor`	Record Descriptor class for defining a Record type and its fields.
`RecordField`
`JsonRecordPacker`
`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.
`RecordPacker`
`RecordPrinter`	Records are printed as textual representation (repr) to fp.
`RecordStreamReader`
`RecordStreamWriter`	Records are written as binary (serialized) to fp.

Functions#

`DynamicDescriptor`
`RecordAdapter`
`RecordReader`
`RecordWriter`
`extend_record`	Extend `record` with fields and values from `other_records`.
`iter_timestamped_records`	Yields timestamped annotated records for each datetime fieldtype in record.
`open_path`	Open path using mode and returns a file object.
`stream`
`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#

`RECORD_VERSION`
`dynamic_fieldtype`

flow.record.RECORD_VERSION = 1#

flow.record.DynamicDescriptor(name, fields)#

class flow.record.FieldType#

classmethod default()#: Return the default value for the field in the Record template.

class flow.record.GroupedRecord(name, records)#

Bases: Record

GroupedRecord acts like a normal Record, but can contain multiple records.

See it as a flat Record view on top of multiple Records. If two Records have the same fieldname, the first one will prevail.

get_record_by_type(type_name)#

Get record in a GroupedRecord by type_name.

Parameters:: type_name (str) – The record type name (for example wq/meta).
Returns:: None or the record

__repr__()#: Return repr(self).

__setattr__(attr, val)#: Enforce setting the fields to their respective types.

__getattr__(attr)#

class flow.record.Record#

__slots__ = ()#

__eq__(other)#: Return self==value.

__setattr__(k, v)#: Enforce setting the fields to their respective types.

__repr__()#: Return repr(self).

flow.record.RecordAdapter(url, out, selector=None, clobber=True, **kwargs)#

class flow.record.RecordDescriptor(name: str, fields: Sequence[Tuple[str, str]] | None = None)#

Record Descriptor class for defining a Record type and its fields.

property fields: Mapping[str, RecordField]#

Get fields mapping (without required fields). eg:

{: “foo”: RecordField(“foo”, “string”), “bar”: RecordField(“bar”, “varint”),

}

Returns:: Mapping of Record fields

property descriptor_hash: int#: Returns the (cached) descriptor hash

property identifier: Tuple[str, int]#: Returns a tuple containing the descriptor name and hash

name: str#

recordType: type#

static get_required_fields() → Mapping[str, RecordField]#

Get required fields mapping. eg:

{: “_source”: RecordField(“_source”, “string”), “_classification”: RecordField(“_classification”, “datetime”), “_generated”: RecordField(“_generated”, “datetime”), “_version”: RecordField(“_version”, “vaeint”),

}

Returns:: Mapping of required fields

get_all_fields() → Mapping[str, RecordField]#

Get all fields including required meta fields. eg:

{: “ts”: RecordField(“ts”, “datetime”), “foo”: RecordField(“foo”, “string”), “bar”: RecordField(“bar”, “varint”), “_source”: RecordField(“_source”, “string”), “_classification”: RecordField(“_classification”, “datetime”), “_generated”: RecordField(“_generated”, “datetime”), “_version”: RecordField(“_version”, “vaeint”),

}

Returns:: Mapping of all Record fields

getfields(typename: str) → RecordFieldSet#

Get fields of a given type.

Parameters:: typename – The typename of the fields to return. eg: “string” or “datetime”
Returns:: RecordFieldSet of fields with the given typename

__call__(*args, **kwargs) → Record#: Create a new Record initialized with args and kwargs.

init_from_dict(rdict: Dict[str, Any], raise_unknown=False) → Record#

Create a new Record initialized with key, value pairs from rdict.

If raise_unknown=True then fields on rdict that are unknown to this RecordDescriptor will raise a TypeError exception due to initializing with unknown keyword arguments. (default: False)

Returns:: Record with data from rdict

init_from_record(record: Record, raise_unknown=False) → Record#

Create a new Record initialized with data from another record.

If raise_unknown=True then fields on record that are unknown to this RecordDescriptor will raise a TypeError exception due to initializing with unknown keyword arguments. (default: False)

Returns:: Record with data from record

extend(fields: Sequence[Tuple[str, str]]) → RecordDescriptor#

Returns a new RecordDescriptor with the extended fields

Returns:: RecordDescriptor with extended fields

get_field_tuples() → Tuple[Tuple[str, str]]#

Returns a tuple containing the (typename, name) tuples, eg:

((‘boolean’, ‘foo’), (‘string’, ‘bar’))

Returns:: Tuple of (typename, name) tuples

static calc_descriptor_hash(name, fields: Sequence[Tuple[str, str]]) → int#

Calculate and return the (cached) descriptor hash as a 32 bit integer.

The descriptor hash is the first 4 bytes of the sha256sum of the descriptor name and field names and types.

__hash__() → int#: Return hash(self).

__eq__(other: RecordDescriptor) → bool#: Return self==value.

__repr__() → str#: Return repr(self).

definition(reserved: bool = True) → str#

Return the RecordDescriptor as Python definition string.

If reserved is True it will also return the reserved fields.

Returns:: Descriptor definition string

base(**kwargs_sink)#

exception flow.record.RecordDescriptorError#

Bases: Exception

Raised when there is an error constructing a record descriptor

class flow.record.RecordField(name: str, typename: str)#

name#

typename#

type#

__repr__()#: Return repr(self).

flow.record.RecordReader(url=None, selector=None, **kwargs)#

flow.record.RecordWriter(url=None, clobber=True, **kwargs)#

flow.record.dynamic_fieldtype#

flow.record.extend_record(record: Record, other_records: List[Record], replace: bool = False, name: str | None = None) → Record#

Extend record with fields and values from other_records.

Duplicate fields are ignored in other_records unless replace=True.

Parameters:

record – Initial Record to extend.
other_records – List of Records to use for extending/replacing.
replace – if True, it will replace existing fields and values in record from fields and values from other_records. Last record always wins.
name – rename the RecordDescriptor name to name. Otherwise, use name from initial record.

Returns:

Extended Record

flow.record.iter_timestamped_records(record: Record) → Iterator[Record]#

Yields timestamped annotated records for each datetime fieldtype in record. If record does not have any datetime fields the original record is returned.

Parameters:: record – Record to add timestamp fields for.
Yields:: Record annotated with ts and ts_description fields for each datetime fieldtype.

flow.record.open_path(path, mode, clobber=True)#

Open path using mode and returns a file object.

It handles special cases if path is meant to be stdin or stdout. And also supports compression based on extension or file header of stream.

Parameters:

path (str) – Filename or path to filename to open
mode (str) – Could be “r”, “rb” to open file for reading, “w”, “wb” for writing
clobber (bool) – Overwrite file if it already exists if clobber=True, else raises IOError.

flow.record.stream(src, dst)#

class flow.record.JsonRecordPacker(indent=None, pack_descriptors=True)#

register(desc, notify=False)#

pack_obj(obj)#

unpack_obj(obj)#

pack(obj)#

unpack(d)#

class flow.record.PathTemplateWriter(path_template=None, name=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'#

rotate_existing_file(path)#

record_stream_for_path(path)#

write(record)#

close()#

class flow.record.RecordArchiver(archive_path, path_template=None, name=None)#

Bases: PathTemplateWriter

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

flow.record.RecordOutput(fp)#: Return a RecordPrinter if fp is a tty otherwise a RecordStreamWriter.

class flow.record.RecordPacker#

EXT_TYPE#

TYPES#

register(desc, notify=False)#

pack_obj(obj, unversioned=False)#

pack(obj)#

unpack_obj(t, data)#

unpack(d)#

class flow.record.RecordPrinter(fp, flush=True)#

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

fp#

write(obj)#

flush()#

close()#

class flow.record.RecordStreamReader(fp, selector=None)#

fp#

recordtype#

descs#

packer#

readheader()#

read()#

close()#

__iter__()#

class flow.record.RecordStreamWriter(fp)#

Records are written as binary (serialized) to fp.

fp#

packer#

__del__()#

on_new_descriptor(descriptor)#

close()#

flush()#

write(obj)#

writeheader()#

flow.record.record_stream(sources, selector=None)#

Return a Record stream generator from the given Record sources.

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

flow.record#

Subpackages#