OpensearchApi has 63 functions (exceeds 20 allowed). Consider refactoring.

Open

class OpensearchApi:
    service = "opensearch"
    version = "2021-01-01"

    @handler("AcceptInboundConnection")

File aws_s3_bucket.py has 536 lines of code (exceeds 250 allowed). Consider refactoring.

Open

# LocalStack Resource Provider Scaffolding v2
from __future__ import annotations

import re
from pathlib import Path

Function determine_aws_service_model has a Cognitive Complexity of 56 (exceeds 5 allowed). Consider refactoring.

Open

def determine_aws_service_model(
    request: Request, services: ServiceCatalog = None
) -> Optional[ServiceModel]:
    """
    Tries to determine the name of the AWS service an incoming request is targeting.

• Read up
  Read up

File connect.py has 529 lines of code (exceeds 250 allowed). Consider refactoring.

Open

"""
LocalStack client stack.

This module provides the interface to perform cross-service communication between
LocalStack providers.

File ASLIntrinsicParser.py has 517 lines of code (exceeds 250 allowed). Consider refactoring.

Open

# Generated from ASLIntrinsicParser.g4 by ANTLR 4.13.1
# encoding: utf-8
from antlr4 import *
from io import StringIO
import sys

File provider.py has 511 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import dataclasses
import json
import logging
import os
import re

File plugins.py has 510 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import abc
import functools
import logging
import threading
from collections import defaultdict

File models.py has 507 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import base64
import datetime
import io
import json
import logging

EventsApi has 57 functions (exceeds 20 allowed). Consider refactoring.

Open

class EventsApi:
    service = "events"
    version = "2015-10-07"

    @handler("ActivateEventSource")

DynamodbApi has 57 functions (exceeds 20 allowed). Consider refactoring.

Open

class DynamodbApi:
    service = "dynamodb"
    version = "2012-08-10"

    @handler("BatchExecuteStatement")

File lambda_models.py has 498 lines of code (exceeds 250 allowed). Consider refactoring.

Open

"""Lambda models for internal use and persistence.
The LambdaProviderPro in localstack-ext imports this model and configures persistence.
The actual function code is stored in S3 (see S3Code).
"""

Function do_run_cmd has a Cognitive Complexity of 51 (exceeds 5 allowed). Consider refactoring.

Open

    def do_run_cmd(self):
        def convert_line(line):
            line = to_str(line or "")
            if self.strip_color:
                # strip color codes

• Read up
  Read up

Function subscribe has a Cognitive Complexity of 49 (exceeds 5 allowed). Consider refactoring.

Open

    def subscribe(
        self,
        context: RequestContext,
        topic_arn: topicARN,
        protocol: String,

• Read up
  Read up

File resource_provider.py has 482 lines of code (exceeds 250 allowed). Consider refactoring.

Open

from __future__ import annotations

import copy
import logging
import re

File api_utils.py has 481 lines of code (exceeds 250 allowed). Consider refactoring.

Open

"""Utilities related to Lambda API operations such as ARN handling, validations, and output formatting.
Everything related to behavior or implicit functionality goes into `lambda_utils.py`.
"""

import datetime

KmsApi has 52 functions (exceeds 20 allowed). Consider refactoring.

Open

class KmsApi:
    service = "kms"
    version = "2014-11-01"

    @handler("CancelKeyDeletion")

EsApi has 51 functions (exceeds 20 allowed). Consider refactoring.

Open

class EsApi:
    service = "es"
    version = "2015-01-01"

    @handler("AcceptInboundCrossClusterSearchConnection")

File __init__.py has 465 lines of code (exceeds 250 allowed). Consider refactoring.

Open

from datetime import datetime
from typing import List, Optional, TypedDict

from localstack.aws.api import RequestContext, ServiceException, ServiceRequest, handler

Function does_match_event has a Cognitive Complexity of 46 (exceeds 5 allowed). Consider refactoring.

Open

def does_match_event(event_pattern: dict[str, any], event: dict[str, any]) -> bool:
    """Decides whether an event pattern matches an event or not.
    Returns True if the `event_pattern` matches the given `event` and False otherwise.

    Implements "Amazon EventBridge event patterns":

• Read up
  Read up

File provider.py has 459 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import datetime
import json
import logging
import os
import re