:py:mod:`flow.record.adapter.elastic`
=====================================

.. py:module:: flow.record.adapter.elastic




Module Contents
---------------

Classes
~~~~~~~

.. autoapisummary::

   flow.record.adapter.elastic.ElasticWriter
   flow.record.adapter.elastic.ElasticReader



Functions
~~~~~~~~~

.. autoapisummary::
   :nosignatures:

   flow.record.adapter.elastic.enrich_elastic_exception



Attributes
~~~~~~~~~~

.. autoapisummary::

   flow.record.adapter.elastic.HAS_ELASTIC
   flow.record.adapter.elastic.__usage__
   flow.record.adapter.elastic.log


.. py:data:: HAS_ELASTIC
   :value: True


.. py:data:: __usage__
   :value: Multiline-String

   .. raw:: html

      <details><summary>Show Value</summary>

   .. code-block:: python

      """
      ElasticSearch adapter
      ---
      Write usage: rdump -w elastic+[PROTOCOL]://[IP]:[PORT]?index=[INDEX]
      Read usage: rdump elastic+[PROTOCOL]://[IP]:[PORT]?index=[INDEX]
      [IP]:[PORT]: ip and port to elastic host
      [PROTOCOL]: http or https. Defaults to https when "+[PROTOCOL]" is omitted
      
      Optional arguments:
        [API_KEY]: base64 encoded api key to authenticate with (default: False)
        [QUEUE_SIZE]: maximum queue size for writing records; limits memory usage (default: 100000)
        [INDEX]: name of the index to use (default: records)
        [VERIFY_CERTS]: verify certs of Elasticsearch instance (default: True)
        [HASH_RECORD]: make record unique by hashing record [slow] (default: False)
        [REQUEST_TIMEOUT]: maximum duration in seconds for a request to Elastic (default: 30)
        [MAX_RETRIES]: maximum retries before a record is marked as failed (default: 3)
        [_META_*]: record metadata fields (default: None)
      """

   .. raw:: html

      </details>



.. py:data:: log

.. py:class:: ElasticWriter(uri: str, index: str = 'records', verify_certs: str | bool = True, http_compress: str | bool = True, hash_record: str | bool = False, api_key: str | None = None, queue_size: int = 100000, request_timeout: int = 30, max_retries: int = 3, **kwargs)

   Bases: :py:obj:`flow.record.adapter.AbstractWriter`


   .. py:attribute:: index
      :value: 'records'



   .. py:attribute:: uri


   .. py:attribute:: hash_record
      :value: False



   .. py:attribute:: max_retries
      :value: 3



   .. py:attribute:: queue
      :type:  queue.Queue[flow.record.base.Record | StopIteration]


   .. py:attribute:: event


   .. py:attribute:: exception
      :type:  Exception | None
      :value: None



   .. py:attribute:: es


   .. py:attribute:: json_packer


   .. py:attribute:: thread


   .. py:attribute:: metadata_fields


   .. py:method:: excepthook(exc: threading.ExceptHookArgs, *args, **kwargs) -> None


   .. py:method:: record_to_document(record: flow.record.base.Record, index: str) -> dict

      Convert a record to a Elasticsearch compatible document dictionary



   .. py:method:: document_stream() -> collections.abc.Iterator[dict]

      Generator of record documents on the Queue



   .. py:method:: streaming_bulk_thread() -> None

      Thread that streams the documents to ES via the bulk api.

      Resources:
          - https://elasticsearch-py.readthedocs.io/en/v8.17.1/helpers.html#elasticsearch.helpers.streaming_bulk
          - https://github.com/elastic/elasticsearch-py/blob/main/elasticsearch/helpers/actions.py#L362



   .. py:method:: write(record: flow.record.base.Record) -> None

      Write a record.



   .. py:method:: flush() -> None

      Flush any buffered writes.



   .. py:method:: close() -> None

      Close the Writer, no more writes will be possible.



.. py:class:: ElasticReader(uri: str, index: str = 'records', verify_certs: str | bool = True, http_compress: str | bool = True, selector: None | flow.record.selector.Selector | flow.record.selector.CompiledSelector = None, api_key: str | None = None, request_timeout: int = 30, max_retries: int = 3, **kwargs)

   Bases: :py:obj:`flow.record.adapter.AbstractReader`


   .. py:attribute:: index
      :value: 'records'



   .. py:attribute:: uri


   .. py:attribute:: selector
      :value: None



   .. py:attribute:: es


   .. py:method:: __iter__() -> collections.abc.Iterator[flow.record.base.Record]

      Return a record iterator.



   .. py:method:: close() -> None

      Close the Reader, can be overriden to properly free resources.



.. py:function:: enrich_elastic_exception(exception: Exception) -> Exception

   Extend the exception with error information from Elastic.

   Resources:
       - https://elasticsearch-py.readthedocs.io/en/v8.17.1/exceptions.html