# Luna Streams v.0.6.2 documentation

## Database context

class luna_streams.db.context.DBContext(logger)

  DB context

  async autorestartStreams()

    Autorestart streams. Make stream.status failure->`pending` for all failed streams that allow autorestart taking restart delay into account Make stream.restart.current_attempt to 0 (means autorestart has been completed) for all streams that have status different from failure after delay seconds and current_attempt >= 1

    Parameters:

      lock – active db connection

    Return type:

      `int`

    Returns:

      amount of the restarted streams

  async blockStreams(connection, streamIds, accountId=None)

    Block streams for update.

    Parameters:

      - connection – current connection

      - streamIds (`list`[`str`]) – stream ids

      - accountId (`Optional`[`str`, `None`]) – account id

    Return type:

      `list`[`str`]

    Returns:

      ids of streams blocked

  async checkStreamAllowedForUpdate(connection, streamId, newStatus=None)

    Check stream is allowed for update with new status :param connection: connection to database :type streamId: `str` :param streamId: stream id :type newStatus: `Optional`[`str`, `None`] :param newStatus: new status for stream

    Raises:

      - VLException(Error.UnableToStopProcessing.format(streamId), 400, False) if unable to stop stream –

      - VLException(Error.UnableToCancelProcessing.format(streamId), 400, False) if unable to cancel stream –

  async countGroups(groupNames=None, groupIds=None, accountId=None)

    Get count of groups.

    Parameters:

      - groupNames (`Optional`[`list`[`str`], `None`]) – group names

      - groupIds (`Optional`[`list`[`str`], `None`]) – group ids

      - accountId (`Optional`[`str`, `None`]) – account id

    Return type:

      `int`

    Returns:

      count of groups

  async countStreams(filters)

    Count streams according to filters.

    Parameters:

      filters (`StreamSearchFilters`) – stream search filters

    Return type:

      `int`

    Returns:

      streams count

  async createGroup(group)

    Create group.

    Parameters:

      group (`NewGroupModel`) – group to create

    Return type:

      `str`

    Returns:

      unique group id

  async createStream(stream, streamId=None, version=1, dbConnection=None)

    Save stream.

    Parameters:

      - stream (`NewStreamModel`) – stream to save

      - streamId (`Optional`[`str`, `None`]) – optional id to put stream by

      - version (`int`) – optional stream version

      - dbConnection (`Optional`[`AbstractDBConnection`, `None`]) – stream database connection, if need to do something within the transaction

    Return type:

      `str`

    Returns:

      stream id in uuid4 format

  async deleteStreamsByDeleteFilters(filters, accountId=None)

    Delete streams by filters (OR).

    Parameters:

      - filters (`DeleteStreamsFilters`) – streams delete filters

      - accountId (`Optional`[`str`, `None`]) – account id

  async deleteStreamsLogs(logTimeLt=None)

    Delete streams logs except the last one :param : logTimeLt: upper excluding bound of stream log creation time

    Return type:

      `int`

    Returns:

      deleted logs count

  async static denyStreamAutoRestart(connection, streamId)

    Deny stream auto restart :type connection: `AbstractDBConnection` :param connection: connection :type streamId: `str` :param streamId: stream id

  async downgradeStreamStatus(streamInactiveTime)

    Downgrade stream status. Make stream.status in_progress->`pending` for all long-time-ago updated streams.

    Parameters:

      streamInactiveTime – period without feedback

    Return type:

      `int`

    Returns:

      amount of the downgraded streams

  getGeoPosition(geoPosition)

    Get geo position prepared for insertion to database :type geoPosition: `dict` :param geoPosition: raw dict with latitude and longitude

    Return type:

      `Union`[`Geometry`, ~T_SDO_GEOMETRY]

    Returns:

      Geometry object for postgres database or SDO_GEOMETRY object for oracle database

  async getGroups(groupNames=None, groupIds=None, accountId=None, page=1, pageSize=None)

    Get groups.

    Parameters:

      - groupNames (`Optional`[`list`[`str`], `None`]) – group names

      - groupIds (`Optional`[`list`[`str`], `None`]) – group ids

      - accountId (`Optional`[`str`, `None`]) – account id

      - page (`int`) – page number

      - pageSize (`Optional`[`int`, `None`]) – page size

    Return type:

      `list`[`dict`[`str`, `Any`]]

    Returns:

      list of groups

  async getPreview(streamId, previewType, accountId=None)

    Get preview url :type streamId: `str` :param streamId: stream id :type previewType: `Literal`[‘last_frame’, ‘live’] :param previewType: preview type - frame or live :type accountId: `Optional`[`str`, `None`] :param accountId: account id

    Return type:

      `str`

    Returns:

      preview url

    Raises:

      - VLException(Error.PreviewNotFound.format(previewType, streamId), 404, False) if preview not found –

      - VLException(Error.StreamNotFound.format(streamId), 404, False) if stream not found –

  async getStreams(filters, page=None, pageSize=None)

    Get streams.

    Parameters:

      - filters (`StreamSearchFilters`) – stream search filters

      - page (`Optional`[`int`, `None`]) – page number

      - pageSize (`Optional`[`int`, `None`]) – page size

    Return type:

      `list`[`dict`[`str`, `Any`]]

    Returns:

      list of streams

  async getStreamsLogs(page, pageSize, targets=None, accountId=None, streamIds=None, statuses=None, logTimeLt=None, logTimeGte=None)

    Get streams logs by filters :type page: `int` :param page: page :type pageSize: `int` :param pageSize: page size :type targets: `Optional`[`list`[`str`], `None`] :param targets: targets :type accountId: `Optional`[`str`, `None`] :param accountId: account id :type streamIds: `Optional`[`list`[`int`], `None`] :param streamIds: stream ids :type statuses: `Optional`[`list`[`str`], `None`] :param statuses: stream statuses :type logTimeLt: `Optional`[`datetime`, `None`] :param logTimeLt: upper excluding bound of stream log creation time :type logTimeGte: `Optional`[`datetime`, `None`] :param logTimeGte: lower including bound of stream log creation time

    Return type:

      `list`[`dict`[`str`, `Any`]]

  async getStreamsQueue(limit, streamIds=None, streamNames=None, groupIds=None, groupNames=None)

    Get streams from the queue.

    Parameters:

      - limit (`int`) – the maximum amount to return

      - streamIds (`Optional`[`list`[`str`], `None`]) – list of stream ids

      - streamNames (`Optional`[`list`[`str`], `None`]) – list of stream names

      - groupIds (`Optional`[`list`[`str`], `None`]) – list of group ids

      - groupNames (`Optional`[`list`[`str`], `None`]) – list of group names

    Return type:

      `list`[`dict`[`str`, `Any`]]

    Returns:

      streams as dicts

  async classmethod initSDO()

    Initialize SDO_GEOMETRY to improve insertion speed | Required only for oracle database See cx_oracle samples/insert_geometry.py for details

    Return type:

      `None`

  async linkStreamsToGroup(streamIds, groupId, accountId=None)

    Attach streamIds to group.

    Parameters:

      - streamIds (`list`[`str`]) – stream ids

      - groupId (`str`) – group id

      - accountId (`Optional`[`str`, `None`]) – account id

    Raises:

      ContextException(Error.StreamNotFound.format(streamId)) if stream doesnt exist –

  static prepareSearchQueryFilters(filters)

    Prepare search query filters on “Stream” model.

    Parameters:

      filters (`StreamSearchFilters`) – query filters

    Return type:

      `BooleanClauseList`

    Returns:

      filters for select query

  async pullStreamsQueue(limit, streamIds=None, streamNames=None, groupIds=None, groupNames=None)

    Pull streams from the queue.

    Parameters:

      - limit (`int`) – the maximum amount to return

      - streamIds (`Optional`[`list`[`str`], `None`]) – list of stream id

      - streamNames (`Optional`[`list`[`str`], `None`]) – list of stream names

      - groupNames (`Optional`[`list`[`str`], `None`]) – list of group names

      - groupIds (`Optional`[`list`[`str`], `None`]) – list of group ids

    Return type:

      `list`[`dict`[`str`, `Any`]]

    Returns:

      streams as dicts

  async putStream(streamId, stream)

    Put stream. There are 3 cases: 1) Replacing existing stream 2) Creating new stream 3) Recreating new stream

    Parameters:

      - streamId (`str`) – id to put stream by

      - stream (`NewStreamModel`) – stream to put

    Raises:

      VLException(Error.StreamNotFound.format(streamId), 404, False) if the stream was deleted –

    Return type:

      `int`

    Returns:

      stream current version

  async removeGroup(groupId, accountId=None)

    Remove group.

    Parameters:

      - groupId (`str`) – group id

      - accountId (`Optional`[`str`, `None`]) – account id

    Return type:

      `int`

    Returns:

      deleted group count

  async removeStreams(streamIds, accountId=None)

    Remove streams by streamIds.

    Parameters:

      - streamIds (`list`[`str`]) – stream ids

      - accountId (`Optional`[`str`, `None`]) – account id

    Return type:

      `int`

    Returns:

      removed streams count

  async saveFeedback(streams)

    Save feedback, return stream info.

    Parameters:

      streams (`list`[`OneStreamFeedback`]) – list of stream feedbacks

    Returns:

      {“modified”: [{<stream info>}], “unmodified”: [{<stream info>}]}

    Return type:

      two lists of db stream info objects

  async unlinkStreamsFromGroup(streamIds, groupId, accountId=None)

    Unlink streamIds from group.

    Parameters:

      - streamIds (`list`[`str`]) – stream ids

      - groupId (`str`) – group id

      - accountId (`Optional`[`str`, `None`]) – account id

  async updateGroup(groupId, group, accountId=None)

    Update group.

    Parameters:

      - groupId (`str`) – group id

      - group (`PatchGroupModel`) – group to update

      - accountId (`Optional`[`str`, `None`]) – account id

    Return type:

      `int`

    Returns:

      updated group count

  async updateStream(streamId, streamData)

    Update stream.

    Parameters:

      - streamId (`str`) – stream id

      - streamData (`StreamPatchingDataModel`) – stream data to update

    Return type:

      `bool`

    Returns:

      True if stream was patched

class luna_streams.db.context.SingleProcessLock(logger, name)

  Single-process lock realisation.

  logger

    logger

  name

    lock name (SingleProcessLock.name)

  context

    database context

  currentDBTimestamp

    sql function for current database calculating

  async heartbeat(name, ttl, sessionId)

    Master heartbeat.

    Parameters:

      - sessionId (`str`) – current lock session

      - name (`str`) – lock name

      - ttl (`int`) – lock ttl

    Return type:

      `bool`

    Returns:

      true if lock has same session id otherwise false

  async lockProcess(name, ttl, sessionId)

    Try to get a lock.

    Lock can be to get if:

      1. if expiration is null

      2. if expiration in the past

    Parameters:

      - name (`str`) – lock name

      - ttl (`int`) – lock ttl

      - sessionId (`str`) – new lock session id

    Return type:

      `bool`

    Returns:

      True if lock is free and we get it (set up expiration time) otherwise False