tracim.lib.core.content

Complexity

Total Complexity

207

Size/Duplication

Total Lines	1250
Duplicated Lines	0 %

Importance

Changes

0

2 Functions

Rating	Name	Duplication	Size	Complexity
B	compare_content_for_sorting_by_type_and_name()	0	37	6
A	compare_tree_items_for_sorting_by_type_and_name()	0	5	1

# -*- coding: utf-8 -*-

This module should have a docstring.

Too many lines in module (1250/1000)

from contextlib import contextmanager

import os

from operator import itemgetter

import transaction
from sqlalchemy import func
from sqlalchemy.orm import Query

__author__ = 'damien'

import datetime

standard import "import datetime" should be placed before "import transaction"

import re

standard import "import re" should be placed before "import transaction"

import typing

standard import "import typing" should be placed before "import transaction"

from tracim.lib.utils.translation import fake_translator as _

from depot.manager import DepotManager
from depot.io.utils import FileIntent

import sqlalchemy

Imports from package sqlalchemy are not grouped

from sqlalchemy.orm import aliased
from sqlalchemy.orm import joinedload
from sqlalchemy.orm.attributes import get_history
from sqlalchemy.orm.session import Session
from sqlalchemy import desc
from sqlalchemy import distinct
from sqlalchemy import or_
from sqlalchemy.sql.elements import and_
from tracim.lib.utils.utils import cmp_to_key

Imports from package tracim are not grouped

from tracim.lib.core.notifications import NotifierFactory
from tracim.exceptions import SameValueError
from tracim.lib.utils.utils import current_date_for_filename
from tracim.models.revision_protection import new_revision
from tracim.models.auth import User
from tracim.models.data import ActionDescription
from tracim.models.data import ContentStatus
from tracim.models.data import ContentRevisionRO
from tracim.models.data import Content
from tracim.models.data import ContentType
from tracim.models.data import NodeTreeItem
from tracim.models.data import RevisionReadStatus
from tracim.models.data import UserRoleInWorkspace
from tracim.models.data import Workspace


def compare_content_for_sorting_by_type_and_name(

The name compare_content_for_sorting_by_type_and_name does not conform to the function naming conventions ((([a-z][a-z0-9_]{2,30})|(_[a-z0-9_]*))$).

        content1: Content,
        content2: Content
) -> int:
    """
    :param content1:
    :param content2:
    :return:    1 if content1 > content2
                -1 if content1 < content2
                0 if content1 = content2
    """

    if content1.type == content2.type:
        if content1.get_label().lower()>content2.get_label().lower():

Exactly one space required around comparison

            return 1
        elif content1.get_label().lower()<content2.get_label().lower():

Exactly one space required around comparison

            return -1
        return 0
    else:
        # TODO - D.A. - 2014-12-02 - Manage Content Types Dynamically

TODO and FIXME comments should generally be avoided.

        content_type_order = [
            ContentType.Folder,
            ContentType.Page,
            ContentType.Thread,
            ContentType.File,
        ]

        content_1_type_index = content_type_order.index(content1.type)
        content_2_type_index = content_type_order.index(content2.type)
        result = content_1_type_index - content_2_type_index

        if result < 0:

Unnecessary "else" after "return"

            return -1
        elif result > 0:
            return 1
        else:
            return 0


def compare_tree_items_for_sorting_by_type_and_name(

The name compare_tree_items_for_sorting_by_type_and_name does not conform to the function naming conventions ((([a-z][a-z0-9_]{2,30})|(_[a-z0-9_]*))$).

This function should have a docstring.

        item1: NodeTreeItem,
        item2: NodeTreeItem
) -> int:
    return compare_content_for_sorting_by_type_and_name(item1.node, item2.node)


class ContentApi(object):

This class should have a docstring.

The variable __class__ seems to be unused.

Too many public methods (50/20)

Too many instance attributes (10/7)

    SEARCH_SEPARATORS = ',| '
    SEARCH_DEFAULT_RESULT_NB = 50

    DISPLAYABLE_CONTENTS = (
        ContentType.Folder,
        ContentType.File,
        ContentType.Comment,
        ContentType.Thread,
        ContentType.Page,
    )

    def __init__(

Too many arguments (10/5)

            self,
            session: Session,
            current_user: typing.Optional[User],

The variable typing does not seem to be defined.

            config,
            show_archived: bool = False,
            show_deleted: bool = False,
            show_temporary: bool = False,
            all_content_in_treeview: bool = True,
            force_show_all_types: bool = False,
            disable_user_workspaces_filter: bool = False,
    ) -> None:
        self._session = session
        self._user = current_user
        self._config = config
        self._user_id = current_user.user_id if current_user else None
        self._show_archived = show_archived
        self._show_deleted = show_deleted
        self._show_temporary = show_temporary
        self._show_all_type_of_contents_in_treeview = all_content_in_treeview
        self._force_show_all_types = force_show_all_types
        self._disable_user_workspaces_filter = disable_user_workspaces_filter

    @contextmanager
    def show(
            self,
            show_archived: bool=False,

Exactly one space required around keyword argument assignment

            show_deleted: bool=False,

Exactly one space required around keyword argument assignment

            show_temporary: bool=False,

Exactly one space required around keyword argument assignment

    ) -> typing.Generator['ContentApi', None, None]:
        """
        Use this method as context manager to update show_archived,
        show_deleted and show_temporary properties during context.
        :param show_archived: show archived contents
        :param show_deleted:  show deleted contents
        :param show_temporary:  show temporary contents
        """
        previous_show_archived = self._show_archived
        previous_show_deleted = self._show_deleted
        previous_show_temporary = self._show_temporary

        try:
            self._show_archived = show_archived
            self._show_deleted = show_deleted
            self._show_temporary = show_temporary
            yield self
        finally:
            self._show_archived = previous_show_archived
            self._show_deleted = previous_show_deleted
            self._show_temporary = previous_show_temporary

    def get_revision_join(self) -> sqlalchemy.sql.elements.BooleanClauseList:
        """
        Return the Content/ContentRevision query join condition
        :return: Content/ContentRevision query join condition
        """
        return and_(Content.id == ContentRevisionRO.content_id,
                    ContentRevisionRO.revision_id == self._session.query(
                        ContentRevisionRO.revision_id)
                    .filter(ContentRevisionRO.content_id == Content.id)
                    .order_by(ContentRevisionRO.revision_id.desc())
                    .limit(1)
                    .correlate(Content))

    def get_canonical_query(self) -> Query:
        """
        Return the Content/ContentRevision base query who join these table on the last revision.
        :return: Content/ContentRevision Query
        """
        return self._session.query(Content)\
            .join(ContentRevisionRO, self.get_revision_join())

    @classmethod
    def sort_tree_items(

This method should have a docstring.

        cls,

Wrong hanging indentation before block (add 4 spaces).

        content_list: typing.List[NodeTreeItem],

The variable typing does not seem to be defined.

Wrong hanging indentation before block (add 4 spaces).

    )-> typing.List[NodeTreeItem]:
        news = []
        for item in content_list:
            news.append(item)

        content_list.sort(key=cmp_to_key(
            compare_tree_items_for_sorting_by_type_and_name,
        ))

        return content_list

    @classmethod
    def sort_content(

This method should have a docstring.

        cls,

Wrong hanging indentation before block (add 4 spaces).

        content_list: typing.List[Content],

The variable typing does not seem to be defined.

Wrong hanging indentation before block (add 4 spaces).

    ) -> typing.List[Content]:
        content_list.sort(key=cmp_to_key(compare_content_for_sorting_by_type_and_name))
        return content_list

    def __real_base_query(
        self,

Wrong hanging indentation before block (add 4 spaces).

        workspace: Workspace = None,

Wrong hanging indentation before block (add 4 spaces).

    ) -> Query:
        result = self.get_canonical_query()

        # Exclude non displayable types
        if not self._force_show_all_types:
            result = result.filter(Content.type.in_(self.DISPLAYABLE_CONTENTS))

        if workspace:
            result = result.filter(Content.workspace_id == workspace.workspace_id)

        # Security layer: if user provided, filter
        # with user workspaces privileges
        if self._user and not self._disable_user_workspaces_filter:
            user = self._session.query(User).get(self._user_id)
            # Filter according to user workspaces
            workspace_ids = [r.workspace_id for r in user.roles \
                             if r.role>=UserRoleInWorkspace.READER]

Exactly one space required around comparison

            result = result.filter(or_(
                Content.workspace_id.in_(workspace_ids),
                # And allow access to non workspace document when he is owner
                and_(
                    Content.workspace_id == None,

Comparison to None should be 'expr is None'

                    Content.owner_id == self._user_id,
                )
            ))

        return result

    def _base_query(self, workspace: Workspace=None) -> Query:

Exactly one space required around keyword argument assignment

        result = self.__real_base_query(workspace)

        if not self._show_deleted:
            result = result.filter(Content.is_deleted==False)

Exactly one space required around comparison

Comparison to False should be 'not expr' or 'expr is False'

        if not self._show_archived:
            result = result.filter(Content.is_archived==False)

Exactly one space required around comparison

Comparison to False should be 'not expr' or 'expr is False'

        if not self._show_temporary:
            result = result.filter(Content.is_temporary==False)

Exactly one space required around comparison

Comparison to False should be 'not expr' or 'expr is False'

        return result

    def __revisions_real_base_query(
        self,

Wrong hanging indentation before block (add 4 spaces).

        workspace: Workspace=None,

Exactly one space required around keyword argument assignment

Wrong hanging indentation before block (add 4 spaces).

    ) -> Query:
        result = self._session.query(ContentRevisionRO)

        # Exclude non displayable types
        if not self._force_show_all_types:
            result = result.filter(Content.type.in_(self.DISPLAYABLE_CONTENTS))

        if workspace:
            result = result.filter(ContentRevisionRO.workspace_id==workspace.workspace_id)

Exactly one space required around comparison

        if self._user:
            user = self._session.query(User).get(self._user_id)
            # Filter according to user workspaces
            workspace_ids = [r.workspace_id for r in user.roles \
                             if r.role>=UserRoleInWorkspace.READER]

Exactly one space required around comparison

            result = result.filter(ContentRevisionRO.workspace_id.in_(workspace_ids))

        return result

    def _revisions_base_query(
        self,

Wrong hanging indentation before block (add 4 spaces).

        workspace: Workspace=None,

Exactly one space required around keyword argument assignment

Wrong hanging indentation before block (add 4 spaces).

    ) -> Query:
        result = self.__revisions_real_base_query(workspace)

        if not self._show_deleted:
            result = result.filter(ContentRevisionRO.is_deleted==False)

Exactly one space required around comparison

Comparison to False should be 'not expr' or 'expr is False'

        if not self._show_archived:
            result = result.filter(ContentRevisionRO.is_archived==False)

Exactly one space required around comparison

Comparison to False should be 'not expr' or 'expr is False'

        if not self._show_temporary:
            result = result.filter(Content.is_temporary==False)

Exactly one space required around comparison

Comparison to False should be 'not expr' or 'expr is False'

        return result

    def _hard_filtered_base_query(
        self,

Wrong hanging indentation before block (add 4 spaces).

        workspace: Workspace=None,

Exactly one space required around keyword argument assignment

Wrong hanging indentation before block (add 4 spaces).

    ) -> Query:
        """
        If set to True, then filterign on is_deleted and is_archived will also
        filter parent properties. This is required for search() function which
        also search in comments (for example) which may be 'not deleted' while
        the associated content is deleted

        :param hard_filtering:
        :return:
        """
        result = self.__real_base_query(workspace)

        if not self._show_deleted:
            parent = aliased(Content)
            result = result.join(parent, Content.parent).\
                filter(Content.is_deleted==False).\

Exactly one space required around comparison

Comparison to False should be 'not expr' or 'expr is False'

                filter(parent.is_deleted==False)

Exactly one space required around comparison

Comparison to False should be 'not expr' or 'expr is False'

        if not self._show_archived:
            parent = aliased(Content)
            result = result.join(parent, Content.parent).\
                filter(Content.is_archived==False).\

Exactly one space required around comparison

Comparison to False should be 'not expr' or 'expr is False'

                filter(parent.is_archived==False)

Exactly one space required around comparison

Comparison to False should be 'not expr' or 'expr is False'

        if not self._show_temporary:
            parent = aliased(Content)
            result = result.join(parent, Content.parent). \
                filter(Content.is_temporary == False). \

Comparison to False should be 'not expr' or 'expr is False'

                filter(parent.is_temporary == False)

Comparison to False should be 'not expr' or 'expr is False'

        return result

    def get_base_query(

This method should have a docstring.

        self,

Wrong hanging indentation before block (add 4 spaces).

        workspace: Workspace,

Wrong hanging indentation before block (add 4 spaces).

    ) -> Query:
        return self._base_query(workspace)

    def get_child_folders(self, parent: Content=None, workspace: Workspace=None, filter_by_allowed_content_types: list=[], removed_item_ids: list=[], allowed_node_types=None) -> typing.List[Content]:

This line is too long as per the coding-style (199/100).

Exactly one space required around keyword argument assignment

The default value [] might cause unintended side-effects.

Too many arguments (6/5)

        """
        This method returns child items (folders or items) for left bar treeview.

        :param parent:
        :param workspace:
        :param filter_by_allowed_content_types:
        :param removed_item_ids:
        :param allowed_node_types: This parameter allow to hide folders for which the given type of content is not allowed.

This line is too long as per the coding-style (123/100).

               For example, if you want to move a Page from a folder to another, you should show only folders that accept pages

This line is too long as per the coding-style (127/100).

        :return:
        """
        filter_by_allowed_content_types = filter_by_allowed_content_types or []  # FDV
        removed_item_ids = removed_item_ids or []  # FDV

        if not allowed_node_types:
            allowed_node_types = [ContentType.Folder]
        elif allowed_node_types==ContentType.Any:

Exactly one space required around comparison

            allowed_node_types = ContentType.all()

        parent_id = parent.content_id if parent else None
        folders = self._base_query(workspace).\
            filter(Content.parent_id==parent_id).\

Exactly one space required around comparison

            filter(Content.type.in_(allowed_node_types)).\
            filter(Content.content_id.notin_(removed_item_ids)).\
            all()

        if not filter_by_allowed_content_types or \
                        len(filter_by_allowed_content_types)<=0:

Exactly one space required around comparison

            # Standard case for the left treeview: we want to show all contents
            # in the left treeview... so we still filter because for example
            # comments must not appear in the treeview
            return [folder for folder in folders \
                    if folder.type in ContentType.allowed_types_for_folding()]

        # Now this is a case of Folders only (used for moving content)
        # When moving a content, you must get only folders that allow to be filled
        # with the type of content you want to move
        result = []
        for folder in folders:
            for allowed_content_type in filter_by_allowed_content_types:

                is_folder = folder.type == ContentType.Folder
                content_type__allowed = folder.properties['allowed_content'][allowed_content_type] == True

This line is too long as per the coding-style (106/100).

Comparison to True should be just 'expr' or 'expr is True'

                if is_folder and content_type__allowed:
                    result.append(folder)
                    break

        return result

    def create(self, content_type: str, workspace: Workspace, parent: Content=None, label:str ='', do_save=False, is_temporary: bool=False) -> Content:

This line is too long as per the coding-style (151/100).

Exactly one space required around keyword argument assignment

Exactly one space required after :

Exactly one space required after keyword argument assignment

This method should have a docstring.

Too many arguments (7/5)

        assert content_type in ContentType.allowed_types()

        if content_type == ContentType.Folder and not label:
            label = self.generate_folder_label(workspace, parent)

        content = Content()
        content.owner = self._user
        content.parent = parent
        content.workspace = workspace
        content.type = content_type
        content.label = label
        content.is_temporary = is_temporary
        content.revision_type = ActionDescription.CREATION

        if content.type in (
                ContentType.Page,
                ContentType.Thread,
        ):
            content.file_extension = '.html'

        if do_save:
            self._session.add(content)
            self.save(content, ActionDescription.CREATION)
        return content


    def create_comment(self, workspace: Workspace=None, parent: Content=None, content:str ='', do_save=False) -> Content:

This line is too long as per the coding-style (121/100).

Exactly one space required around keyword argument assignment

Exactly one space required after :

Exactly one space required after keyword argument assignment

This method should have a docstring.

        assert parent  and parent.type!=ContentType.Folder

Exactly one space required around comparison

        item = Content()
        item.owner = self._user
        item.parent = parent
        item.workspace = workspace
        item.type = ContentType.Comment
        item.description = content
        item.label = ''
        item.revision_type = ActionDescription.COMMENT

        if do_save:
            self.save(item, ActionDescription.COMMENT)
        return item


    def get_one_from_revision(self, content_id: int, content_type: str, workspace: Workspace=None, revision_id=None) -> Content:

This line is too long as per the coding-style (128/100).

Exactly one space required around keyword argument assignment

        """
        This method is a hack to convert a node revision item into a node
        :param content_id:
        :param content_type:
        :param workspace:
        :param revision_id:
        :return:
        """

        content = self.get_one(content_id, content_type, workspace)
        revision = self._session.query(ContentRevisionRO).filter(ContentRevisionRO.revision_id==revision_id).one()

This line is too long as per the coding-style (114/100).

Exactly one space required around comparison

        if revision.content_id==content.content_id:

Exactly one space required around comparison

            content.revision_to_serialize = revision.revision_id
        else:
            raise ValueError('Revision not found for given content')

        return content

    def get_one(self, content_id: int, content_type: str, workspace: Workspace=None) -> Content:

Exactly one space required around keyword argument assignment

This method should have a docstring.

        if not content_id:
            return None

        if content_type==ContentType.Any:

Exactly one space required around comparison

            return self._base_query(workspace).filter(Content.content_id==content_id).one()

Exactly one space required around comparison

        return self._base_query(workspace).filter(Content.content_id==content_id).filter(Content.type==content_type).one()

This line is too long as per the coding-style (122/100).

Exactly one space required around comparison

    def get_one_revision(self, revision_id: int = None) -> ContentRevisionRO:
        """
        This method allow us to get directly any revision with its id
        :param revision_id: The content's revision's id that we want to return
        :return: An item Content linked with the correct revision
        """
        assert revision_id is not None# DYN_REMOVE

        revision = self._session.query(ContentRevisionRO).filter(ContentRevisionRO.revision_id == revision_id).one()

This line is too long as per the coding-style (116/100).

        return revision

    # INFO - A.P - 2017-07-03 - python file object getter
    # in case of we cook a version of preview manager that allows a pythonic
    # access to files
    # def get_one_revision_file(self, revision_id: int = None):
    #     """
    #     This function allows us to directly get a Python file object from its
    #     revision identifier.
    #     :param revision_id: The revision id of the file we want to return
    #     :return: The corresponding Python file object
    #     """
    #     revision = self.get_one_revision(revision_id)
    #     return DepotManager.get().get(revision.depot_file)

    def get_one_revision_filepath(self, revision_id: int = None) -> str:
        """
        This method allows us to directly get a file path from its revision
        identifier.
        :param revision_id: The revision id of the filepath we want to return
        :return: The corresponding filepath
        """
        revision = self.get_one_revision(revision_id)
        depot = DepotManager.get()
        depot_stored_file = depot.get(revision.depot_file)  # type: StoredFile
        depot_file_path = depot_stored_file._file_path  # type: str

It seems like _file_path was declared protected and should not be accessed from this context.

        return depot_file_path

    def get_one_by_label_and_parent(
            self,
            content_label: str,
            content_parent: Content=None,

Exactly one space required around keyword argument assignment

    ) -> Content:
        """
        This method let us request the database to obtain a Content with its name and parent
        :param content_label: Either the content's label or the content's filename if the label is None

This line is too long as per the coding-style (103/100).

        :param content_parent: The parent's content
        :param workspace: The workspace's content
        :return The corresponding Content
        """
        workspace = content_parent.workspace if content_parent else None
        query = self._base_query(workspace)
        parent_id = content_parent.content_id if content_parent else None
        query = query.filter(Content.parent_id == parent_id)

        file_name, file_extension = os.path.splitext(content_label)

        return query.filter(
            or_(
                and_(
                    Content.type == ContentType.File,
                    Content.label == file_name,
                    Content.file_extension == file_extension,
                ),
                and_(
                    Content.type == ContentType.Thread,
                    Content.label == file_name,
                ),
                and_(
                    Content.type == ContentType.Page,
                    Content.label == file_name,
                ),
                and_(
                    Content.type == ContentType.Folder,
                    Content.label == content_label,
                ),
            )
        ).one()

    def get_one_by_label_and_parent_labels(

The name get_one_by_label_and_parent_labels does not conform to the method naming conventions ((([a-z][a-z0-9_]{2,30})|(_[a-z0-9_]*))$).

            self,
            content_label: str,
            workspace: Workspace,
            content_parent_labels: [str]=None,

Exactly one space required around keyword argument assignment

    ):
        """
        Return content with it's label, workspace and parents labels (optional)
        :param content_label: label of content (label or file_name)
        :param workspace: workspace containing all of this
        :param content_parent_labels: Ordered list of labels representing path
            of folder (without workspace label).
        E.g.: ['foo', 'bar'] for complete path /Workspace1/foo/bar folder
        :return: Found Content
        """
        query = self._base_query(workspace)
        parent_folder = None

        # Grab content parent folder if parent path given
        if content_parent_labels:
            parent_folder = self.get_folder_with_workspace_path_labels(
                content_parent_labels,
                workspace,
            )

        # Build query for found content by label
        content_query = self.filter_query_for_content_label_as_path(
            query=query,
            content_label_as_file=content_label,
        )

        # Modify query to apply parent folder filter if any
        if parent_folder:
            content_query = content_query.filter(
                Content.parent_id == parent_folder.content_id,
            )
        else:
            content_query = content_query.filter(
                Content.parent_id == None,

Comparison to None should be 'expr is None'

            )

        # Filter with workspace
        content_query = content_query.filter(
            Content.workspace_id == workspace.workspace_id,
        )

        # Return the content
        return content_query\
            .order_by(
                Content.revision_id.desc(),
            )\
            .one()

    def get_folder_with_workspace_path_labels(

The name get_folder_with_workspace_path_labels does not conform to the method naming conventions ((([a-z][a-z0-9_]{2,30})|(_[a-z0-9_]*))$).

            self,
            path_labels: [str],
            workspace: Workspace,
    ) -> Content:
        """
        Return a Content folder for given relative path.
        TODO BS 20161124: Not safe if web interface allow folder duplicate names
        :param path_labels: List of labels representing path of folder
        (without workspace label).
        E.g.: ['foo', 'bar'] for complete path /Workspace1/foo/bar folder
        :param workspace: workspace of folders
        :return: Content folder
        """
        query = self._base_query(workspace)
        folder = None

        for label in path_labels:
            # Filter query on label
            folder_query = query \
                .filter(
                    Content.type == ContentType.Folder,
                    Content.label == label,
                    Content.workspace_id == workspace.workspace_id,
                )

            # Search into parent folder (if already deep)
            if folder:
                folder_query = folder_query\
                    .filter(
                        Content.parent_id == folder.content_id,
                    )
            else:
                folder_query = folder_query \
                    .filter(Content.parent_id == None)

Comparison to None should be 'expr is None'

            # Get thirst corresponding folder
            folder = folder_query \
                .order_by(Content.revision_id.desc()) \
                .one()

        return folder

    def filter_query_for_content_label_as_path(

The name filter_query_for_content_label_as_path does not conform to the method naming conventions ((([a-z][a-z0-9_]{2,30})|(_[a-z0-9_]*))$).

This method could be written as a function/class method.

            self,
            query: Query,
            content_label_as_file: str,
            is_case_sensitive: bool = False,
    ) -> Query:
        """
        Apply normalised filters to found Content corresponding as given label.
        :param query: query to modify
        :param content_label_as_file: label in this
        FILE version, use Content.get_label_as_file().
        :param is_case_sensitive: Take care about case or not
        :return: modified query
        """
        file_name, file_extension = os.path.splitext(content_label_as_file)

        label_filter = Content.label == content_label_as_file
        file_name_filter = Content.label == file_name
        file_extension_filter = Content.file_extension == file_extension

        if not is_case_sensitive:
            label_filter = func.lower(Content.label) == \
                           func.lower(content_label_as_file)
            file_name_filter = func.lower(Content.label) == \
                               func.lower(file_name)
            file_extension_filter = func.lower(Content.file_extension) == \
                                    func.lower(file_extension)

        return query.filter(or_(
            and_(
                Content.type == ContentType.File,
                file_name_filter,
                file_extension_filter,
            ),
            and_(
                Content.type == ContentType.Thread,
                file_name_filter,
                file_extension_filter,
            ),
            and_(
                Content.type == ContentType.Page,
                file_name_filter,
                file_extension_filter,
            ),
            and_(
                Content.type == ContentType.Folder,
                label_filter,
            ),
        ))

    def get_all(self, parent_id: int=None, content_type: str=ContentType.Any, workspace: Workspace=None) -> typing.List[Content]:

This line is too long as per the coding-style (129/100).

Exactly one space required around keyword argument assignment

This method should have a docstring.

        assert parent_id is None or isinstance(parent_id, int) # DYN_REMOVE
        assert content_type is not None# DYN_REMOVE
        assert isinstance(content_type, str) # DYN_REMOVE

        resultset = self._base_query(workspace)

        if content_type!=ContentType.Any:

Exactly one space required around comparison

            resultset = resultset.filter(Content.type==content_type)

Exactly one space required around comparison

        if parent_id:
            resultset = resultset.filter(Content.parent_id==parent_id)

Exactly one space required around comparison

        if parent_id is False:
            resultset = resultset.filter(Content.parent_id == None)

Comparison to None should be 'expr is None'

        return resultset.all()

    def get_children(self, parent_id: int, content_types: list, workspace: Workspace=None) -> typing.List[Content]:

This line is too long as per the coding-style (115/100).

Exactly one space required around keyword argument assignment

        """
        Return parent_id childs of given content_types
        :param parent_id: parent id
        :param content_types: list of types
        :param workspace: workspace filter
        :return: list of content
        """
        resultset = self._base_query(workspace)
        resultset = resultset.filter(Content.type.in_(content_types))

        if parent_id:
            resultset = resultset.filter(Content.parent_id==parent_id)

Exactly one space required around comparison

        if parent_id is False:
            resultset = resultset.filter(Content.parent_id == None)

Comparison to None should be 'expr is None'

        return resultset.all()

    # TODO find an other name to filter on is_deleted / is_archived

TODO and FIXME comments should generally be avoided.

    def get_all_with_filter(self, parent_id: int=None, content_type: str=ContentType.Any, workspace: Workspace=None) -> typing.List[Content]:

This line is too long as per the coding-style (141/100).

Exactly one space required around keyword argument assignment

This method should have a docstring.

        assert parent_id is None or isinstance(parent_id, int) # DYN_REMOVE
        assert content_type is not None# DYN_REMOVE
        assert isinstance(content_type, str) # DYN_REMOVE

        resultset = self._base_query(workspace)

        if content_type != ContentType.Any:
            resultset = resultset.filter(Content.type==content_type)

Exactly one space required around comparison

        resultset = resultset.filter(Content.is_deleted == self._show_deleted)
        resultset = resultset.filter(Content.is_archived == self._show_archived)
        resultset = resultset.filter(Content.is_temporary == self._show_temporary)

        resultset = resultset.filter(Content.parent_id==parent_id)

Exactly one space required around comparison

        return resultset.all()

    def get_all_without_exception(self, content_type: str, workspace: Workspace=None) -> typing.List[Content]:

This line is too long as per the coding-style (110/100).

Exactly one space required around keyword argument assignment

This method should have a docstring.

        assert content_type is not None# DYN_REMOVE

        resultset = self._base_query(workspace)

        if content_type != ContentType.Any:
            resultset = resultset.filter(Content.type==content_type)

Exactly one space required around comparison

        return resultset.all()

    def get_last_active(self, parent_id: int, content_type: str, workspace: Workspace=None, limit=10) -> typing.List[Content]:

This line is too long as per the coding-style (126/100).

Exactly one space required around keyword argument assignment

This method should have a docstring.

        assert parent_id is None or isinstance(parent_id, int) # DYN_REMOVE
        assert content_type is not None# DYN_REMOVE
        assert isinstance(content_type, str) # DYN_REMOVE

        resultset = self._base_query(workspace) \
            .filter(Content.workspace_id == Workspace.workspace_id) \
            .filter(Workspace.is_deleted.is_(False)) \
            .order_by(desc(Content.updated))

        if content_type!=ContentType.Any:

Exactly one space required around comparison

            resultset = resultset.filter(Content.type==content_type)

Exactly one space required around comparison

        if parent_id:
            resultset = resultset.filter(Content.parent_id==parent_id)

Exactly one space required around comparison

        result = []
        for item in resultset:
            new_item = None
            if ContentType.Comment == item.type:
                new_item = item.parent
            else:
                new_item = item

            # INFO - D.A. - 2015-05-20
            # We do not want to show only one item if the last 10 items are
            # comments about one thread for example
            if new_item not in result:
                result.append(new_item)

            if len(result) >= limit:
                break

        return result

    def get_last_unread(self, parent_id: int, content_type: str,

This method should have a docstring.

                        workspace: Workspace=None, limit=10) -> typing.List[Content]:

Exactly one space required around keyword argument assignment

        assert parent_id is None or isinstance(parent_id, int) # DYN_REMOVE
        assert content_type is not None# DYN_REMOVE
        assert isinstance(content_type, str) # DYN_REMOVE

        read_revision_ids = self._session.query(RevisionReadStatus.revision_id) \
            .filter(RevisionReadStatus.user_id==self._user_id)

Exactly one space required around comparison

        not_read_revisions = self._revisions_base_query(workspace) \
            .filter(~ContentRevisionRO.revision_id.in_(read_revision_ids)) \
            .filter(ContentRevisionRO.workspace_id == Workspace.workspace_id) \
            .filter(Workspace.is_deleted.is_(False)) \
            .subquery()

        not_read_content_ids_query = self._session.query(
            distinct(not_read_revisions.c.content_id)
        )
        not_read_content_ids = list(map(
            itemgetter(0),
            not_read_content_ids_query,
        ))

        not_read_contents = self._base_query(workspace) \
            .filter(Content.content_id.in_(not_read_content_ids)) \
            .order_by(desc(Content.updated))

        if content_type != ContentType.Any:
            not_read_contents = not_read_contents.filter(
                Content.type==content_type)

Exactly one space required around comparison

        else:
            not_read_contents = not_read_contents.filter(
                Content.type!=ContentType.Folder)

Exactly one space required around comparison

        if parent_id:
            not_read_contents = not_read_contents.filter(
                Content.parent_id==parent_id)

Exactly one space required around comparison

        result = []
        for item in not_read_contents:
            new_item = None
            if ContentType.Comment == item.type:
                new_item = item.parent
            else:
                new_item = item

            # INFO - D.A. - 2015-05-20
            # We do not want to show only one item if the last 10 items are
            # comments about one thread for example
            if new_item not in result:
                result.append(new_item)

            if len(result) >= limit:
                break

        return result

    def set_allowed_content(self, folder: Content, allowed_content_dict:dict):

Exactly one space required after :

This method could be written as a function/class method.

        """
        :param folder: the given folder instance
        :param allowed_content_dict: must be something like this:
            dict(
                folder = True
                thread = True,
                file = False,
                page = True
            )
        :return:
        """
        properties = dict(allowed_content = allowed_content_dict)

No space allowed around keyword argument assignment

        folder.properties = properties

    def set_status(self, content: Content, new_status: str):

This method should have a docstring.

This method could be written as a function/class method.

        if new_status in ContentStatus.allowed_values():
            content.status = new_status
            content.revision_type = ActionDescription.STATUS_UPDATE
        else:
            raise ValueError('The given value {} is not allowed'.format(new_status))

    def move(self, item: Content,

This method should have a docstring.

This method could be written as a function/class method.

             new_parent: Content,
             must_stay_in_same_workspace:bool=True,

Exactly one space required after :

Exactly one space required around keyword argument assignment

             new_workspace:Workspace=None):

Exactly one space required after :

Exactly one space required around keyword argument assignment

        if must_stay_in_same_workspace:
            if new_parent and new_parent.workspace_id != item.workspace_id:
                raise ValueError('the item should stay in the same workspace')

        item.parent = new_parent
        if new_parent:
            item.workspace = new_parent.workspace
        elif new_workspace:
            item.workspace = new_workspace

        item.revision_type = ActionDescription.MOVE

    def copy(

Too many arguments (6/5)

        self,

Wrong hanging indentation before block (add 4 spaces).

        item: Content,

Wrong hanging indentation before block (add 4 spaces).

        new_parent: Content=None,

Exactly one space required around keyword argument assignment

Wrong hanging indentation before block (add 4 spaces).

        new_label: str=None,

Exactly one space required around keyword argument assignment

Wrong hanging indentation before block (add 4 spaces).

        do_save: bool=True,

Exactly one space required around keyword argument assignment

Wrong hanging indentation before block (add 4 spaces).

        do_notify: bool=True,

Exactly one space required around keyword argument assignment

Wrong hanging indentation before block (add 4 spaces).

    ) -> Content:
        """
        Copy nearly all content, revision included. Children not included, see
        "copy_children" for this.
        :param item: Item to copy
        :param new_parent: new parent of the new copied item
        :param new_label: new label of the new copied item
        :param do_notify: notify copy or not
        :return: Newly copied item
        """
        if (not new_parent and not new_label) or (new_parent == item.parent and new_label == item.label):  # nopep8

This line is too long as per the coding-style (115/100).

            # TODO - G.M - 08-03-2018 - Use something else than value error

TODO and FIXME comments should generally be avoided.

            raise ValueError("You can't copy file into itself")
        if new_parent:
            workspace = new_parent.workspace
            parent = new_parent
        else:
            workspace = item.workspace
            parent = item.parent
        label = new_label or item.label

        content = item.copy(parent)
        # INFO - GM - 15-03-2018 - add "copy" revision
        with new_revision(
            session=self._session,
            tm=transaction.manager,
            content=content,
            force_create_new_revision=True
        ) as rev:
            rev.parent = parent
            rev.workspace = workspace
            rev.label = label
            rev.revision_type = ActionDescription.COPY
            rev.properties['origin'] = {
                'content': item.id,
                'revision': item.last_revision.revision_id,
            }
        if do_save:
            self.save(content, ActionDescription.COPY, do_notify=do_notify)
        return content

    def copy_children(self, origin_content: Content, new_content: Content):

This method should have a docstring.

        for child in origin_content.children:
            self.copy(child, new_content)

    def move_recursively(self, item: Content,

This method should have a docstring.

                         new_parent: Content, new_workspace: Workspace):
        self.move(item, new_parent, False, new_workspace)
        self.save(item, do_notify=False)

        for child in item.children:
            with new_revision(child):

It seems like a value for argument tm is missing in the function call.

It seems like a value for argument content is missing in the function call.

                self.move_recursively(child, item, new_workspace)
        return

    def update_content(self, item: Content, new_label: str, new_content: str=None) -> Content:

Exactly one space required around keyword argument assignment

This method should have a docstring.

        if item.label==new_label and item.description==new_content:

Exactly one space required around comparison

            # TODO - G.M - 20-03-2018 - Fix internatization for webdav access.

TODO and FIXME comments should generally be avoided.

            # Internatization disabled in libcontent for now.
            raise SameValueError('The content did not changed')
        item.owner = self._user
        item.label = new_label
        item.description = new_content if new_content else item.description # TODO: convert urls into links

TODO and FIXME comments should generally be avoided.

This line is too long as per the coding-style (107/100).

        item.revision_type = ActionDescription.EDITION
        return item

    def update_file_data(self, item: Content, new_filename: str, new_mimetype: str, new_content: bytes) -> Content:

This line is too long as per the coding-style (115/100).

This method should have a docstring.

        if new_mimetype == item.file_mimetype and \
                new_content == item.depot_file.file.read():
            raise SameValueError('The content did not changed')
        item.owner = self._user
        item.file_name = new_filename
        item.file_mimetype = new_mimetype
        item.depot_file = FileIntent(
            new_content,
            new_filename,
            new_mimetype,
        )
        item.revision_type = ActionDescription.REVISION
        return item

    def archive(self, content: Content):

This method should have a docstring.

        content.owner = self._user
        content.is_archived = True
        # TODO - G.M - 12-03-2018 - Inspect possible label conflict problem

TODO and FIXME comments should generally be avoided.

        # INFO - G.M - 12-03-2018 - Set label name to avoid trouble when
        # un-archiving file.
        content.label = '{label}-{action}-{date}'.format(
            label=content.label,
            action='archived',
            date=current_date_for_filename()
        )
        content.revision_type = ActionDescription.ARCHIVING

    def unarchive(self, content: Content):

This method should have a docstring.

        content.owner = self._user
        content.is_archived = False
        content.revision_type = ActionDescription.UNARCHIVING

    def delete(self, content: Content):

This method should have a docstring.

        content.owner = self._user
        content.is_deleted = True
        # TODO - G.M - 12-03-2018 - Inspect possible label conflict problem

TODO and FIXME comments should generally be avoided.

        # INFO - G.M - 12-03-2018 - Set label name to avoid trouble when
        # un-deleting file.
        content.label = '{label}-{action}-{date}'.format(
            label=content.label,
            action='deleted',
            date=current_date_for_filename()
        )
        content.revision_type = ActionDescription.DELETION

    def undelete(self, content: Content):

This method should have a docstring.

        content.owner = self._user
        content.is_deleted = False
        content.revision_type = ActionDescription.UNDELETION

    def mark_read__all(self,

This method should have a docstring.

                       read_datetime: datetime=None,

Exactly one space required around keyword argument assignment

                       do_flush: bool=True,

Exactly one space required around keyword argument assignment

                       recursive: bool=True

Exactly one space required around keyword argument assignment

                       ):

Wrong continued indentation (remove 1 space).

        itemset = self.get_last_unread(None, ContentType.Any)

        for item in itemset:
            self.mark_read(item, read_datetime, do_flush, recursive)

    def mark_read__workspace(self,

This method should have a docstring.

                       workspace : Workspace,

Wrong continued indentation (add 6 spaces).

No space allowed before :

                       read_datetime: datetime=None,

Wrong continued indentation (add 6 spaces).

Exactly one space required around keyword argument assignment

                       do_flush: bool=True,

Wrong continued indentation (add 6 spaces).

Exactly one space required around keyword argument assignment

                       recursive: bool=True

Wrong continued indentation (add 6 spaces).

Exactly one space required around keyword argument assignment

                       ):

Wrong continued indentation (add 5 spaces).

        itemset = self.get_last_unread(None, ContentType.Any, workspace)

        for item in itemset:
            self.mark_read(item, read_datetime, do_flush, recursive)

    def mark_read(self, content: Content,

This method should have a docstring.

                  read_datetime: datetime=None,

Exactly one space required around keyword argument assignment

                  do_flush: bool=True, recursive: bool=True) -> Content:

Exactly one space required around keyword argument assignment

        assert self._user
        assert content

        # The algorithm is:
        # 1. define the read datetime
        # 2. update all revisions related to current Content
        # 3. do the same for all child revisions
        #    (ie parent_id is content_id of current content)

        if not read_datetime:
            read_datetime = datetime.datetime.now()

        viewed_revisions = self._session.query(ContentRevisionRO) \
            .filter(ContentRevisionRO.content_id==content.content_id).all()

Exactly one space required around comparison

        for revision in viewed_revisions:
            revision.read_by[self._user] = read_datetime

        if recursive:
            # mark read :
            # - all children
            # - parent stuff (if you mark a comment as read,
            #                 then you have seen the parent)
            # - parent comments
            for child in content.get_valid_children():
                self.mark_read(child, read_datetime=read_datetime,
                               do_flush=False)

            if ContentType.Comment == content.type:
                self.mark_read(content.parent, read_datetime=read_datetime,
                               do_flush=False, recursive=False)
                for comment in content.parent.get_comments():
                    if comment != content:
                        self.mark_read(comment, read_datetime=read_datetime,
                                       do_flush=False, recursive=False)

        if do_flush:
            self.flush()

        return content

    def mark_unread(self, content: Content, do_flush=True) -> Content:

This method should have a docstring.

        assert self._user
        assert content

        revisions = self._session.query(ContentRevisionRO) \
            .filter(ContentRevisionRO.content_id==content.content_id).all()

Exactly one space required around comparison

        for revision in revisions:
            del revision.read_by[self._user]

        for child in content.get_valid_children():
            self.mark_unread(child, do_flush=False)

        if do_flush:
            self.flush()

        return content

    def flush(self):

This method should have a docstring.

        self._session.flush()

    def save(self, content: Content, action_description: str=None, do_flush=True, do_notify=True):

Exactly one space required around keyword argument assignment

        """
        Save an object, flush the session and set the revision_type property
        :param content:
        :param action_description:
        :return:
        """
        assert action_description is None or action_description in ActionDescription.allowed_values()

This line is too long as per the coding-style (101/100).

        if not action_description:
            # See if the last action has been modified
            if content.revision_type==None or len(get_history(content.revision, 'revision_type'))<=0:

This line is too long as per the coding-style (101/100).

Exactly one space required around comparison

Comparison to None should be 'expr is None'

                # The action has not been modified, so we set it to default edition
                action_description = ActionDescription.EDITION

        if action_description:
            content.revision_type = action_description

        if do_flush:
            # INFO - 2015-09-03 - D.A.
            # There are 2 flush because of the use
            # of triggers for content creation
            #
            # (when creating a content, actually this is an insert of a new
            # revision in content_revisions ; so the mark_read operation need
            # to get full real data from database before to be prepared.

            self._session.add(content)
            self._session.flush()

            # TODO - 2015-09-03 - D.A. - Do not use triggers

TODO and FIXME comments should generally be avoided.

            # We should create a new ContentRevisionRO object instead of Content
            # This would help managing view/not viewed status
            self.mark_read(content, do_flush=True)

        if do_notify:
            self.do_notify(content)

    def do_notify(self, content: Content):
        """
        Allow to force notification for a given content. By default, it is
        called during the .save() operation
        :param content:
        :return:
        """
        NotifierFactory.create(
            self._config,
            self._user
        ).notify_content_update(content)

    def get_keywords(self, search_string, search_string_separators=None) -> [str]:

This method could be written as a function/class method.

        """
        :param search_string: a list of coma-separated keywords
        :return: a list of str (each keyword = 1 entry
        """

        search_string_separators = search_string_separators or ContentApi.SEARCH_SEPARATORS

        keywords = []
        if search_string:
            keywords = [keyword.strip() for keyword in re.split(search_string_separators, search_string)]

This line is too long as per the coding-style (105/100).

        return keywords

    def search(self, keywords: [str]) -> Query:
        """
        :return: a sorted list of Content items
        """

        if len(keywords)<=0:

Exactly one space required around comparison

            return None

        filter_group_label = list(Content.label.ilike('%{}%'.format(keyword)) for keyword in keywords)

This line is too long as per the coding-style (102/100).

        filter_group_desc = list(Content.description.ilike('%{}%'.format(keyword)) for keyword in keywords)

This line is too long as per the coding-style (107/100).

        title_keyworded_items = self._hard_filtered_base_query().\
            filter(or_(*(filter_group_label+filter_group_desc))).\
            options(joinedload('children_revisions')).\
            options(joinedload('parent'))

        return title_keyworded_items

    def get_all_types(self) -> typing.List[ContentType]:

This method should have a docstring.

This method could be written as a function/class method.

        labels = ContentType.all()
        content_types = []
        for label in labels:
            content_types.append(ContentType(label))

        return ContentType.sorted(content_types)

    def exclude_unavailable(
        self,

Wrong hanging indentation before block (add 4 spaces).

        contents: typing.List[Content],

The variable typing does not seem to be defined.

Wrong hanging indentation before block (add 4 spaces).

    ) -> typing.List[Content]:
        """
        Update and return list with content under archived/deleted removed.
        :param contents: List of contents to parse
        """
        for content in contents[:]:
            if self.content_under_deleted(content) or self.content_under_archived(content):
                contents.remove(content)
        return contents

    def content_under_deleted(self, content: Content) -> bool:

This method should have a docstring.

        if content.parent:
            if content.parent.is_deleted:
                return True
            if content.parent.parent:
                return self.content_under_deleted(content.parent)
        return False

    def content_under_archived(self, content: Content) -> bool:

This method should have a docstring.

        if content.parent:
            if content.parent.is_archived:
                return True
            if content.parent.parent:
                return self.content_under_archived(content.parent)
        return False

    def find_one_by_unique_property(
            self,
            property_name: str,
            property_value: str,
            workspace: Workspace=None,

Exactly one space required around keyword argument assignment

    ) -> Content:
        """
        Return Content who contains given property.
        Raise sqlalchemy.orm.exc.MultipleResultsFound if more than one Content
        contains this property value.
        :param property_name: Name of property
        :param property_value: Value of property
        :param workspace: Workspace who contains Content
        :return: Found Content
        """
        # TODO - 20160602 - Bastien: Should be JSON type query

TODO and FIXME comments should generally be avoided.

        # see https://www.compose.io/articles/using-json-extensions-in-\
        # postgresql-from-python-2/
        query = self._base_query(workspace=workspace).filter(
            Content._properties.like(

It seems like _properties was declared protected and should not be accessed from this context.

                '%"{property_name}": "{property_value}"%'.format(
                    property_name=property_name,
                    property_value=property_value,
                )
            )
        )
        return query.one()

    def generate_folder_label(
            self,
            workspace: Workspace,
            parent: Content=None,

Exactly one space required around keyword argument assignment

    ) -> str:
        """
        Generate a folder label
        :param workspace: Future folder workspace
        :param parent: Parent of foture folder (can be None)
        :return: Generated folder name
        """
        query = self._base_query(workspace=workspace)\
            .filter(Content.label.ilike('{0}%'.format(
                _('New folder'),
            )))
        if parent:
            query = query.filter(Content.parent == parent)

        return _('New folder {0}').format(
            query.count() + 1,
        )