Function get_history_for_room has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def get_history_for_room(room_id: str, user_id: str, last_read: str = None) -> list:
    history = environ.env.config.get(
            ConfigKeys.TYPE,
            domain=ConfigKeys.HISTORY,
            default=ConfigKeys.DEFAULT_HISTORY_STRATEGY)

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                            if (v !== undefined) {
                                value[k] = v;
                            } else {
                                delete value[k];
                            }

Function _on_join_emit_join_event has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def _on_join_emit_join_event(arg: tuple) -> None:
    activity = arg[1]

    image = ""
    user_name = None

• Read up
  Read up

Function get_owners has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def get_owners(activity: Activity) -> list:
        if not hasattr(activity.target, 'attachments') or activity.target.attachments is None:
            return list()
        for attachment in activity.target.attachments:
            if not hasattr(attachment, 'object_type'):

• Read up
  Read up

Function remove_sid_for_user has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def remove_sid_for_user(self, user_id: str, sid: str) -> None:
        @with_session
        def update_sid(session=None):
            user_sid = session.query(Sids)\
                .filter(Sids.sid == sid)\

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if not can_whisper:
                            return False, reason_code, 'not allowed to whisper this user'
                    else:

Function ban_user_room has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def ban_user_room(self, user_id: str, ban_timestamp: str, ban_duration: str, room_id: str, reason: str=None, banner_id: str=None):

Function _do_post has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def _do_post(self, room_name, user_ids, owner_id, owner_name, channel_id, temporary: bool = True):

Function __call__ has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def __call__(self, *args, **kwargs):
        activity = args[0]
        # env = args[1]
        # acl_type = args[2]
        # acl_values = args[3]

• Read up
  Read up

Function _validate_params has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def _validate_params(self):
        is_valid, msg, json = self.validate_json(self.request, silent=False)
        if not is_valid:
            raise RuntimeError('invalid json: %s' % msg)

• Read up
  Read up

Function validate_new_acl has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def validate_new_acl(self, values: str):
        if values is None or len(values.strip()) == 0:
            raise ValidationException('blank range when creating AclRangeValidator')

        if ':' not in values:

• Read up
  Read up

Function on_whisper has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def on_whisper(self, activity: Activity) -> (bool, int, str):
        if not hasattr(activity, 'target') or not hasattr(activity.target, 'id'):
            return False, ECodes.MISSING_TARGET_ID, 'no target.id (user uuid to whisper to)'
        if not hasattr(activity, 'actor') or not hasattr(activity.actor, 'id'):
            return False, ECodes.MISSING_ACTOR_ID, 'no actor.id (id of user who is whispering)'

• Read up
  Read up

Function created_activity has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

def created_activity(user_id: str, user_name: str, target_id: str, target_name: str, session_ids: list, namespace: str) -> dict:

Function remove_current_rooms_for_user has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def remove_current_rooms_for_user(self, user_id):
        @with_session
        def remove(room_sids: dict, session=None) -> None:

            for i in range(3):

• Read up
  Read up

Function create_room has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def create_room(
            self, room_name: str, room_id: str, channel_id: str, user_id: str,
            user_name: str, ephemeral: bool=True, sort_order: int=999, is_sid_room=False
    ) -> None:
        @with_session

• Read up
  Read up

Function do_delete has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def do_delete(self):
        try:
            is_valid, msg, json = self.validate_json(self.request, silent=False)
            if not is_valid or 'user_id' not in json:
                raise RuntimeError('invalid json: %s' % msg)

• Read up
  Read up

Function init_logging has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def init_logging(gn_env: GNEnvironment) -> None:
    if len(gn_env.config) == 0 or gn_env.config.get(ConfigKeys.TESTING, False):
        # assume we're testing
        return

• Read up
  Read up

Function check_heartbeats has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def check_heartbeats(self, user_ids: list) -> None:
        for user_id in user_ids:
            still_online = self.env.cache.check_heartbeat(user_id)

            if still_online:

• Read up
  Read up

Function ban_user_room has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def ban_user_room(self, user_id: str, ban_timestamp: str, ban_duration: str, room_id: str, reason: str=None, banner_id: str=None) -> None:

Function ban_user_global has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def ban_user_global(self, user_id: str, ban_timestamp: str, ban_duration: str, reason: str=None, banner_id: str=None, session=None):