File query2.py has 350 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import logging
from abc import abstractmethod
from datetime import datetime
from typing import (
    Any,

File sqlite.py has 329 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import json
import logging
import os
import sqlite3
from datetime import datetime, timedelta, timezone

File peewee.py has 327 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import json
import logging
import os
from datetime import datetime, timedelta, timezone
from typing import (

Function flood has a Cognitive Complexity of 26 (exceeds 5 allowed). Consider refactoring.

Open

def flood(events: List[Event], pulsetime: float = 5) -> List[Event]:
    """
    Takes a list of events and "floods" any empty space between events by extending one of the surrounding events to cover the empty space.

    For more details on flooding, see this issue:

• Read up
  Read up

Function merge_events_by_keys has a Cognitive Complexity of 22 (exceeds 5 allowed). Consider refactoring.

Open

def merge_events_by_keys(events, keys) -> List[Event]:
    """
    Sums the duration of all events which share a value for a key and returns a new event for each value.

    .. note: The result will be a list of events without timestamp since they are merged.

• Read up
  Read up

Function union_no_overlap has a Cognitive Complexity of 22 (exceeds 5 allowed). Consider refactoring.

Open

def union_no_overlap(events1: List[Event], events2: List[Event]) -> List[Event]:
    """Merges two eventlists and removes overlap, the first eventlist will have precedence

    Example:
      events1  | xxx    xx     xxx     |

• Read up
  Read up

Function check has a Cognitive Complexity of 21 (exceeds 5 allowed). Consider refactoring.

Open

    def check(string: str):
        i = 0
        # Find opening bracket
        found = False
        for char in string:

• Read up
  Read up

Function insert has a Cognitive Complexity of 20 (exceeds 5 allowed). Consider refactoring.

Open

    def insert(self, events: Union[Event, List[Event]]) -> Optional[Event]:
        """
        Inserts one or several events.
        If a single event is inserted, return the event with its id assigned.
        If several events are inserted, returns None. (This is due to there being no efficient way of getting ids out when doing bulk inserts with some datastores such as peewee/SQLite)

• Read up
  Read up

Function print_log has a Cognitive Complexity of 16 (exceeds 5 allowed). Consider refactoring.

Open

def print_log(
    path: Path, since: Optional[datetime] = None, level: Optional[str] = None
):
    if not path.is_file():
        return

• Read up
  Read up

Function _intersecting_eventpairs has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

def _intersecting_eventpairs(
    events1: List[Event], events2: List[Event]
) -> Iterable[Tuple[Event, Event, Timeslot]]:
    """A generator that yields each overlapping pair of events from two eventlists along with a Timeslot of the intersection"""
    events1.sort(key=lambda e: e.timestamp)

• Read up
  Read up

Function _merge has a Cognitive Complexity of 12 (exceeds 5 allowed). Consider refactoring.

Open

def _merge(a: dict, b: dict, path=None):
    """
    Recursively merges b into a, with b taking precedence.

    From: https://stackoverflow.com/a/7205107/965332

• Read up
  Read up

Function update_bucket has a Cognitive Complexity of 12 (exceeds 5 allowed). Consider refactoring.

Open

    def update_bucket(
        self,
        bucket_id: str,
        type_id: Optional[str] = None,
        client: Optional[str] = None,

• Read up
  Read up

Function update_bucket has a Cognitive Complexity of 12 (exceeds 5 allowed). Consider refactoring.

Open

    def update_bucket(
        self,
        bucket_id: str,
        type_id: Optional[str] = None,
        client: Optional[str] = None,

• Read up
  Read up

Function union has a Cognitive Complexity of 12 (exceeds 5 allowed). Consider refactoring.

Open

def union(events1: List[Event], events2: List[Event]) -> List[Event]:
    """
    Concatenates and sorts union of 2 event lists and removes duplicates.

    Example:

• Read up
  Read up

Function get_events has a Cognitive Complexity of 12 (exceeds 5 allowed). Consider refactoring.

Open

    def get_events(
        self,
        bucket_id: str,
        limit: int,
        starttime: Optional[datetime] = None,

• Read up
  Read up

Function check has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

    def check(string: str):
        if string[0] != "{":
            return None, string
        # Find closing bracket
        i = 1

• Read up
  Read up

Function check has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

    def check(string: str):
        if string[0] != "[":
            return None, string
        # Find closing bracket
        i = 1

• Read up
  Read up

Function q2_function has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

def q2_function(transform_func=None):
    """
    Decorator used to register query functions.

    Automatically adds mock arguments for Datastore and TNamespace

• Read up
  Read up

Function q2_find_bucket has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

def q2_find_bucket(
    datastore: Datastore, filter_str: str, hostname: Optional[str] = None
):
    """Find bucket by using a filter_str (to avoid hardcoding bucket names)"""
    for bucket in datastore.buckets():

• Read up
  Read up

Consider simplifying this complex logical expression.

Open

            if char == "'" and prev_char != "\\" and not double_quote:
                single_quote = not single_quote
            elif char == '"' and prev_char != "\\" and not single_quote:
                double_quote = not double_quote
            elif single_quote or double_quote: