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

Open

import logging
from typing import Union, List, Dict, Sequence, Callable, Type, Any, Tuple
from datetime import datetime

from aw_core.models import Event

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 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 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

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

Open

from typing import Optional, List, Iterable
from datetime import datetime, timezone, timedelta
import json
import os
import logging

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

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

Open

from typing import Optional, List, Dict, Any
from datetime import datetime, timezone, timedelta
import json
import os
import logging

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

Open

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

    show_levels = LOGLEVELS[LOGLEVELS.index(level) :] if level else None

• 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 insert has a Cognitive Complexity of 13 (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 _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 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 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: str = None):
    """Find bucket by using a filter_str (to avoid hardcoding bucket names)"""
    for bucket in datastore.buckets():
        if filter_str in bucket:
            bucket_metadata = datastore[bucket].metadata()

• 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

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:

Function chunk_events_by_key has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

def chunk_events_by_key(
    events: List[Event], key: str, pulsetime: float = 5.0
) -> List[Event]:
    """
    "Chunks" adjacent events together which have the same value for a key, and stores the

• Read up
  Read up

Function setup_logging has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

def setup_logging(
    name: str,
    testing=False,
    verbose=False,
    log_stderr=True,

• Read up
  Read up