Code clean (annotations, method rename, method move)

tracim/tracim/lib/content.py

                     .correlate(Content))
 
     @classmethod
-    def get_base_query(cls):
+    def get_canonical_query(cls):
         """
         Return the Content/ContentRevision base query who join these table on the last revision.
         :return: Content/ContentRevision Query
         return breadcrumb
 
     def __real_base_query(self, workspace: Workspace=None):
-        result = self.get_base_query()
+        result = self.get_canonical_query()
 
         if workspace:
             result = result.filter(Content.workspace_id==workspace.workspace_id)

tracim/tracim/model/__init__.py

 from decorator import contextmanager
 from sqlalchemy import event, inspect, MetaData
 from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.orm import scoped_session, sessionmaker
-
+from sqlalchemy.orm import scoped_session, sessionmaker, Session
+from sqlalchemy.orm.unitofwork import UOWTransaction
 from zope.sqlalchemy import ZopeTransactionExtension
 
 from tracim.lib.exception import ContentRevisionUpdateError, ContentRevisionDeleteError
     _updatable_revisions = []
 
     @classmethod
-    def add_to_updatable(cls, revision):
+    def add_to_updatable(cls, revision: 'ContentRevisionRO') -> None:
         if inspect(revision).has_identity:
             raise ContentRevisionUpdateError("ContentRevision is not updatable. %s already have identity." % revision)
 
             cls._updatable_revisions.append(revision)
 
     @classmethod
-    def remove_from_updatable(cls, revision):
+    def remove_from_updatable(cls, revision: 'ContentRevisionRO') -> None:
         cls._updatable_revisions.remove(revision)
 
     @classmethod
-    def is_updatable(cls, revision):
+    def is_updatable(cls, revision: 'ContentRevisionRO') -> bool:
         return revision in cls._updatable_revisions
 
 # Global session manager: DBSession() returns the Thread-local
 # defined with SQLAlchemy's declarative extension, but if you need more
 # control, you can switch to the traditional method.
 convention = {
-  "ix": 'ix__%(column_0_label)s',
-  "uq": "uk__%(table_name)s__%(column_0_name)s",
-  "ck": "ck__%(table_name)s__%(constraint_name)s",
-  "fk": "fk__%(table_name)s__%(column_0_name)s__%(referred_table_name)s",
-  "pk": "pk__%(table_name)s"
+  "ix": 'ix__%(column_0_label)s',  # Indexes
+  "uq": "uq__%(table_name)s__%(column_0_name)s",  # Unique constrains
+  "ck": "ck__%(table_name)s__%(constraint_name)s",  # Other column constrains
+  "fk": "fk__%(table_name)s__%(column_0_name)s__%(referred_table_name)s",  # Foreign keys
+  "pk": "pk__%(table_name)s"  # Primary keys
 }
 
 metadata = MetaData(naming_convention=convention)
 
 
 @event.listens_for(DBSession, 'before_flush')
-def prevent_content_revision_delete(session, flush_context, instances):
+def prevent_content_revision_delete(session: Session, flush_context: UOWTransaction,
+                                    instances: [DeclarativeBase]) -> None:
     for instance in session.deleted:
         if isinstance(instance, ContentRevisionRO) and instance.revision_id is not None:
             raise ContentRevisionDeleteError("ContentRevision is not deletable. You must make a new revision with" +
 
 
 @contextmanager
-def new_revision(content):
+def new_revision(content: Content) -> Content:
     """
     Prepare context to update a Content. It will add a new updatable revision to the content.
     :param content: Content instance to update
             if inspect(content.revision).has_identity:
                 content.new_revision()
             RevisionsIntegrity.add_to_updatable(content.revision)
-            yield content.revision
+            yield content
         finally:
             RevisionsIntegrity.remove_from_updatable(content.revision)

tracim/tracim/model/data.py

 from sqlalchemy.orm import backref
 from sqlalchemy.orm import deferred
 from sqlalchemy.orm import relationship
+from sqlalchemy.orm.attributes import InstrumentedAttribute
 from sqlalchemy.orm.collections import attribute_mapped_collection
 from sqlalchemy.types import Boolean
 from sqlalchemy.types import DateTime
     label   = Column(Unicode(1024), unique=False, nullable=False, default='')
     description = Column(Text(), unique=False, nullable=False, default='')
 
+    #  Default value datetime.utcnow, see: http://stackoverflow.com/a/13370382/801924 (or http://pastebin.com/VLyWktUn)
     created = Column(DateTime, unique=False, nullable=False, default=datetime.utcnow)
+    #  Default value datetime.utcnow, see: http://stackoverflow.com/a/13370382/801924 (or http://pastebin.com/VLyWktUn)
     updated = Column(DateTime, unique=False, nullable=False, default=datetime.utcnow)
 
     is_deleted = Column(Boolean, unique=False, nullable=False, default=False)
 
     """ List of column copied when make a new revision from another """
     _cloned_columns = (
-        'content_id', 'owner_id', 'label', 'description', 'file_name', 'file_mimetype', 'properties',
-        'file_content', 'type', 'status', 'created', 'updated', 'is_deleted', 'is_archived',
-        'revision_type', 'workspace_id', 'workspace', 'parent_id', 'parent', 'node', 'owner'
+        'content_id', 'created', 'description', 'file_content', 'file_mimetype', 'file_name', 'is_archived',
+        'is_deleted', 'label', 'node', 'owner' 'owner_id', 'parent', 'parent_id', 'properties', 'revision_type',
+        'status', 'type', 'updated', 'workspace', 'workspace_id',
+    )
+
+    # Read by must be used like this:
+    # read_datetime = revision.ready_by[<User instance>]
+    # if user did not read the content, then a key error is raised
+    read_by = association_proxy(
+        'revision_read_statuses',  # name of the attribute
+        'view_datetime',  # attribute the value is taken from
+        creator=lambda k, v: \
+            RevisionReadStatus(user=k, view_datetime=v)
     )
 
     @classmethod
-    def new_from(cls, revision):
+    def new_from(cls, revision: 'ContentRevisionRO') -> 'ContentRevisionRO':
         """
 
         Return new instance of ContentRevisionRO where properties are copied from revision parameter.
 
         return new_rev
 
-    def __setattr__(self, key, value):
+    def __setattr__(self, key: str, value: 'mixed'):
         """
         ContentRevisionUpdateError is raised if tried to update column and revision own identity
         :param key: attribute name
 
         super().__setattr__(key, value)
 
-    def get_status(self):
+    def get_status(self) -> ContentStatus:
         return ContentStatus(self.status)
 
-    def get_label(self):
+    def get_label(self) -> str:
         return self.label if self.label else self.file_name if self.file_name else ''
 
     def get_last_action(self) -> ActionDescription:
         return ActionDescription(self.revision_type)
 
-    # Read by must be used like this:
-    # read_datetime = revision.ready_by[<User instance>]
-    # if user did not read the content, then a key error is raised
-    read_by = association_proxy(
-        'revision_read_statuses',  # name of the attribute
-        'view_datetime',  # attribute the value is taken from
-        creator=lambda k, v: \
-            RevisionReadStatus(user=k, view_datetime=v)
-    )
-
     def has_new_information_for(self, user: User) -> bool:
         """
         :param user: the session current user
                   .filter(Content.label == 'foo')
                   .one()
 
-    ContentApi provide also prepared Content at tracim.lib.content.ContentApi#get_base_query:
+    ContentApi provide also prepared Content at tracim.lib.content.ContentApi#get_canonical_query:
 
-    content = ContentApi.get_base_query()
+    content = ContentApi.get_canonical_query()
               .filter(Content.label == 'foo')
               .one()
     """
                                       back_populates="parent")
 
     @hybrid_property
-    def content_id(self):
+    def content_id(self) -> int:
         return self.revision.content_id
 
     @content_id.setter
-    def content_id(self, value):
+    def content_id(self, value: int) -> None:
         self.revision.content_id = value
 
     @content_id.expression
-    def content_id(cls):
+    def content_id(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.content_id
 
     @hybrid_property
-    def revision_id(self):
+    def revision_id(self) -> int:
         return self.revision.revision_id
 
     @revision_id.setter
-    def revision_id(self, value):
+    def revision_id(self, value: int):
         self.revision.revision_id = value
 
     @revision_id.expression
-    def revision_id(cls):
+    def revision_id(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.revision_id
 
     @hybrid_property
-    def owner_id(self):
+    def owner_id(self) -> int:
         return self.revision.owner_id
 
     @owner_id.setter
-    def owner_id(self, value):
+    def owner_id(self, value: int) -> None:
         self.revision.owner_id = value
 
     @owner_id.expression
-    def owner_id(cls):
+    def owner_id(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.owner_id
 
     @hybrid_property
-    def label(self):
+    def label(self) -> str:
         return self.revision.label
 
     @label.setter
-    def label(self, value):
+    def label(self, value: str) -> None:
         self.revision.label = value
 
     @label.expression
-    def label(cls):
+    def label(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.label
 
     @hybrid_property
-    def description(self):
+    def description(self) -> str:
         return self.revision.description
 
     @description.setter
-    def description(self, value):
+    def description(self, value: str) -> None:
         self.revision.description = value
 
     @description.expression
-    def description(cls):
+    def description(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.description
 
     @hybrid_property
-    def file_name(self):
+    def file_name(self) -> str:
         return self.revision.file_name
 
     @file_name.setter
-    def file_name(self, value):
+    def file_name(self, value: str) -> None:
         self.revision.file_name = value
 
     @file_name.expression
-    def file_name(cls):
+    def file_name(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.file_name
 
     @hybrid_property
-    def file_mimetype(self):
+    def file_mimetype(self) -> str:
         return self.revision.file_mimetype
 
     @file_mimetype.setter
-    def file_mimetype(self, value):
+    def file_mimetype(self, value: str) -> None:
         self.revision.file_mimetype = value
 
     @file_mimetype.expression
-    def file_mimetype(cls):
+    def file_mimetype(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.file_mimetype
 
     @hybrid_property
         self.revision.file_content = value
 
     @file_content.expression
-    def file_content(cls):
+    def file_content(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.file_content
 
     @hybrid_property
-    def _properties(self):
+    def _properties(self) -> str:
         return self.revision.properties
 
     @_properties.setter
-    def _properties(self, value):
+    def _properties(self, value: str) -> None:
         self.revision.properties = value
 
     @_properties.expression
-    def _properties(cls):
+    def _properties(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.properties
 
     @hybrid_property
-    def type(self):
+    def type(self) -> str:
         return self.revision.type
 
     @type.setter
-    def type(self, value):
+    def type(self, value: str) -> None:
         self.revision.type = value
 
     @type.expression
-    def type(cls):
+    def type(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.type
 
     @hybrid_property
-    def status(self):
+    def status(self) -> str:
         return self.revision.status
 
     @status.setter
-    def status(self, value):
+    def status(self, value: str) -> None:
         self.revision.status = value
 
     @status.expression
-    def status(cls):
+    def status(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.status
 
     @hybrid_property
-    def created(self):
+    def created(self) -> datetime:
         return self.revision.created
 
     @created.setter
-    def created(self, value):
+    def created(self, value: datetime) -> None:
         self.revision.created = value
 
     @created.expression
-    def created(cls):
+    def created(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.created
 
     @hybrid_property
-    def updated(self):
+    def updated(self) -> datetime:
         return self.revision.updated
 
     @updated.setter
-    def updated(self, value):
+    def updated(self, value: datetime) -> None:
         self.revision.updated = value
 
     @updated.expression
-    def updated(cls):
+    def updated(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.updated
 
     @hybrid_property
-    def is_deleted(self):
+    def is_deleted(self) -> bool:
         return self.revision.is_deleted
 
     @is_deleted.setter
-    def is_deleted(self, value):
+    def is_deleted(self, value: bool) -> None:
         self.revision.is_deleted = value
 
     @is_deleted.expression
-    def is_deleted(cls):
+    def is_deleted(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.is_deleted
 
     @hybrid_property
-    def is_archived(self):
+    def is_archived(self) -> bool:
         return self.revision.is_archived
 
     @is_archived.setter
-    def is_archived(self, value):
+    def is_archived(self, value: bool) -> None:
         self.revision.is_archived = value
 
     @is_archived.expression
-    def is_archived(cls):
+    def is_archived(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.is_archived
 
     @hybrid_property
-    def revision_type(self):
+    def revision_type(self) -> str:
         return self.revision.revision_type
 
     @revision_type.setter
-    def revision_type(self, value):
+    def revision_type(self, value: str) -> None:
         self.revision.revision_type = value
 
     @revision_type.expression
-    def revision_type(cls):
+    def revision_type(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.revision_type
 
     @hybrid_property
-    def workspace_id(self):
+    def workspace_id(self) -> int:
         return self.revision.workspace_id
 
     @workspace_id.setter
-    def workspace_id(self, value):
+    def workspace_id(self, value: int) -> None:
         self.revision.workspace_id = value
 
     @workspace_id.expression
-    def workspace_id(cls):
+    def workspace_id(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.workspace_id
 
     @hybrid_property
-    def workspace(self):
+    def workspace(self) -> Workspace:
         return self.revision.workspace
 
     @workspace.setter
-    def workspace(self, value):
+    def workspace(self, value: Workspace) -> None:
         self.revision.workspace = value
 
     @workspace.expression
-    def workspace(cls):
+    def workspace(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.workspace
 
     @hybrid_property
-    def parent_id(self):
+    def parent_id(self) -> int:
         return self.revision.parent_id
 
     @parent_id.setter
-    def parent_id(self, value):
+    def parent_id(self, value: int) -> None:
         self.revision.parent_id = value
 
     @parent_id.expression
-    def parent_id(cls):
+    def parent_id(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.parent_id
 
     @hybrid_property
-    def parent(self):
+    def parent(self) -> 'Content':
         return self.revision.parent
 
     @parent.setter
-    def parent(self, value):
+    def parent(self, value: 'Content') -> None:
         self.revision.parent = value
 
     @parent.expression
-    def parent(cls):
+    def parent(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.parent
 
     @hybrid_property
-    def node(self):
+    def node(self) -> 'Content':
         return self.revision.node
 
     @node.setter
-    def node(self, value):
+    def node(self, value: 'Content') -> None:
         self.revision.node = value
 
     @node.expression
-    def node(cls):
+    def node(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.node
 
     @hybrid_property
-    def owner(self):
+    def owner(self) -> User:
         return self.revision.owner
 
     @owner.setter
-    def owner(self, value):
+    def owner(self, value: User) -> None:
         self.revision.owner = value
 
     @owner.expression
-    def owner(cls):
+    def owner(cls) -> InstrumentedAttribute:
         return ContentRevisionRO.owner
 
     @hybrid_property
-    def children(self):
+    def children(self) -> ['Content']:
         """
         :return: list of children Content
         :rtype Content
         return list(set([revision.node for revision in self.children_revisions]))
 
     @property
-    def revision(self):
+    def revision(self) -> ContentRevisionRO:
         return self.get_current_revision()
 
-    def get_current_revision(self):
+    def get_current_revision(self) -> ContentRevisionRO:
         if not self.revisions:
             return self.new_revision()
 
         revisions = sorted(self.revisions, key=lambda revision: revision.revision_id)
         return revisions[-1]
 
-    def new_revision(self):
+    def new_revision(self) -> None:
         """
         Return and assign to this content a new revision.
         If it's a new content, revision is totally new.
                     yield child.node
 
     @hybrid_property
-    def properties(self):
+    def properties(self) -> dict:
         """ return a structure decoded from json content of _properties """
         if not self._properties:
             ContentChecker.reset_properties(self)
         return json.loads(self._properties)
 
     @properties.setter
-    def properties(self, properties_struct):
+    def properties(self, properties_struct: dict) -> None:
         """ encode a given structure into json and store it in _properties attribute"""
         self._properties = json.dumps(properties_struct)
         ContentChecker.check_properties(self)
 
     revision_id = Column(Integer, ForeignKey('content_revisions.revision_id', ondelete='CASCADE', onupdate='CASCADE'), primary_key=True)
     user_id = Column(Integer, ForeignKey('users.user_id', ondelete='CASCADE', onupdate='CASCADE'), primary_key=True)
-    view_datetime = Column(DateTime, unique=False, nullable=False, server_default=func.now())
+    #  Default value datetime.utcnow, see: http://stackoverflow.com/a/13370382/801924 (or http://pastebin.com/VLyWktUn)
+    view_datetime = Column(DateTime, unique=False, nullable=False, default=datetime.utcnow)
 
     content_revision = relationship(
         'ContentRevisionRO',

tracim/tracim/tests/__init__.py

 
 class BaseTest(object):
 
-    def _create_workspace(self, name, *args, **kwargs):
+    def _create_workspace(self, name, *args, **kwargs) -> Workspace:
         """
         All extra parameters (*args, **kwargs) are for Workspace init
         :return: Created workspace instance
-        :rtype: Workspace
         """
         workspace = Workspace(label=name, *args, **kwargs)
         DBSession.add(workspace)
         eq_(1, DBSession.query(Workspace).filter(Workspace.label == name).count())
         return DBSession.query(Workspace).filter(Workspace.label == name).one()
 
-    def _create_content(self, name, workspace, *args, **kwargs):
+    def _create_content(self, name, workspace, *args, **kwargs) -> Content:
         """
         All extra parameters (*args, **kwargs) are for Content init
         :return: Created Content instance
-        :rtype: Content
         """
         content = Content(*args, **kwargs)
         content.label = name
         DBSession.add(content)
         DBSession.flush()
 
-        eq_(1, ContentApi.get_base_query().filter(Content.label == name).count())
-        return ContentApi.get_base_query().filter(Content.label == name).one()
+        eq_(1, ContentApi.get_canonical_query().filter(Content.label == name).count())
+        return ContentApi.get_canonical_query().filter(Content.label == name).one()
 
 
 class BaseTestThread(BaseTest):
 
-    def _create_thread(self, workspace_name='workspace_1', folder_name='folder_1', thread_name='thread_1'):
+    def _create_thread(self, workspace_name='workspace_1', folder_name='folder_1', thread_name='thread_1') -> Content:
         """
         :return: Thread
-        :rtype: Content
         """
         workspace = self._create_workspace(workspace_name)
         folder = self._create_content(folder_name, workspace, type=ContentType.Folder)

tracim/tracim/tests/library/test_workspace.py

             thread_name='thread_1',
         )
         workspace = DBSession.query(Workspace).filter(Workspace.label == 'workspace_1').one()
-        folder = ContentApi.get_base_query().filter(Content.label == 'folder_1').one()
+        folder = ContentApi.get_canonical_query().filter(Content.label == 'folder_1').one()
         eq_([folder, ], list(workspace.get_valid_children()))

tracim/tracim/tests/models/test_content_revision.py

 
 class TestContentRevision(BaseTest, TestStandard):
 
+    def _new_from(self, revision):
+        excluded_columns = ('revision_id', '_sa_instance_state')
+        revision_columns = [attr.key for attr in inspect(revision).attrs if not attr.key in excluded_columns]
+        new_revision = ContentRevisionRO()
+
+        for revision_column in revision_columns:
+            old_revision_column_value = getattr(revision, revision_column)
+            setattr(new_revision, revision_column, old_revision_column_value)
+
+        return new_revision
+
+    def _get_dict_representation(self, revision):
+        keys_to_remove = ('updated', '_sa_instance_state')
+        dict_repr = OrderedDict(sorted(revision.__dict__.items()))
+        for key_to_remove in keys_to_remove:
+            dict_repr.pop(key_to_remove, None)
+        return dict_repr
+
     def test_new_revision(self):
         admin = DBSession.query(User).filter(User.email == 'admin@admin.admin').one()
         workspace = self._create_workspace(name='workspace_1')
 
         # They must be identical
         eq_(new_revision_by_model_dict, new_revision_by_test_dict)
-
-    def _new_from(self, revision):
-        excluded_columns = ('revision_id', '_sa_instance_state')
-        revision_columns = [attr.key for attr in inspect(revision).attrs if not attr.key in excluded_columns]
-        new_revision = ContentRevisionRO()
-
-        for revision_column in revision_columns:
-            old_revision_column_value = getattr(revision, revision_column)
-            setattr(new_revision, revision_column, old_revision_column_value)
-
-        return new_revision
-
-    def _get_dict_representation(self, revision):
-        keys_to_remove = ('updated', '_sa_instance_state')
-        dict_repr = OrderedDict(sorted(revision.__dict__.items()))
-        for key_to_remove in keys_to_remove:
-            dict_repr.pop(key_to_remove, None)
-        return dict_repr