flow.record
#
Subpackages#
flow.record.adapter
flow.record.adapter.archive
flow.record.adapter.avro
flow.record.adapter.broker
flow.record.adapter.csvfile
flow.record.adapter.elastic
flow.record.adapter.jsonfile
flow.record.adapter.line
flow.record.adapter.mongo
flow.record.adapter.split
flow.record.adapter.splunk
flow.record.adapter.stream
flow.record.adapter.text
flow.record.adapter.xlsx
flow.record.fieldtypes
flow.record.tools
Submodules#
Package Contents#
Classes#
GroupedRecord acts like a normal Record, but can contain multiple records. |
|
Record Descriptor class for defining a Record type and its fields. |
|
Write records to a path on disk, path can be a template string. |
|
RecordWriter that writes/archives records to a path with YYYY/mm/dd. |
|
Records are printed as textual representation (repr) to fp. |
|
Records are written as binary (serialized) to fp. |
Functions#
Extend |
|
Yields timestamped annotated records for each datetime fieldtype in record. |
|
Open path using mode and returns a file object. |
|
Return a RecordPrinter if fp is a tty otherwise a RecordStreamWriter. |
|
Return a Record stream generator from the given Record sources. |
Attributes#
- 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
- 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 fromother_records
.Duplicate fields are ignored in
other_records
unlessreplace=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 inrecord
from fields and values fromother_records
. Last record always wins.name – rename the RecordDescriptor name to
name
. Otherwise, use name from initialrecord
.
- 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.