kallithea Changeset - c8f34ad333a2

Changeset - c8f34ad333a2

Parent rev.

Child rev.

[Not reviewed]

default

0 6 0

Søren Løvborg - 9 years ago 2017-01-18 16:30:13
sorenl@unity3d.com

model: simplify guess_instance invocations

The second argument to the guess_instance classmethod is in practice
fixed for Gist, Permission, RepoGroup, Repository, User, UserGroup;
so might as well move this logic into per-class specializations of
guess_instance.

6 files changed with 33 insertions and 17 deletions:

kallithea/model/base.py

kallithea/model/db.py

kallithea/model/gist.py

kallithea/model/repo.py

kallithea/model/repo_group.py

kallithea/model/user_group.py

0 comments (0 inline, 0 general)

kallithea/model/base.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.base
 ~~~~~~~~~~~~~~~~~~~~
 The application's model objects
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Nov 25, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 from kallithea.model import meta
 from kallithea.lib.utils2 import obfuscate_url_pw
 log = logging.getLogger(__name__)
 def init_model(engine):
     """
     Initializes db session, bind the engine with the metadata,
     Call this before using any of the tables or classes in the model,
     preferably once in application start
     :param engine: engine to bind to
     """
     engine_str = obfuscate_url_pw(str(engine.url))
     log.info("initializing db for %s", engine_str)
     meta.Base.metadata.bind = engine
 class BaseModel(object):
     """
     Base Model for all Kallithea models, it adds sql alchemy session
     into instance of model
     :param sa: If passed it reuses this session instead of creating a new one
     """
     def __init__(self, sa=None):
         if sa is not None:
             self.sa = sa
         else:
             self.sa = meta.Session()
     def _get_user(self, user):
         """
         Helper method to get user by ID, or username fallback
         :param user: UserID, username, or User instance
         """
         from kallithea.model.db import User
         return User.guess_instance(user,
                                   callback=User.get_by_username)
         return User.guess_instance(user)
     def _get_repo(self, repository):
         """
         Helper method to get repository by ID, or repository name
         :param repository: RepoID, repository name or Repository Instance
         """
         from kallithea.model.db import Repository
         return Repository.guess_instance(repository,
                                   callback=Repository.get_by_repo_name)
         return Repository.guess_instance(repository)
     def _get_perm(self, permission):
         """
         Helper method to get permission by ID, or permission name
         :param permission: PermissionID, permission_name or Permission instance
         """
         from kallithea.model.db import Permission
         return Permission.guess_instance(permission,
                                   callback=Permission.get_by_key)
         return Permission.guess_instance(permission)

kallithea/model/db.py

➞

Show inline comments

@@ @@ -371,942 +371,954 @@ class Ui(Base, BaseDbModel): @@
     def get_by_key(cls, section, key):
         """ Return specified Ui object, or None if not found. """
         return cls.query().filter_by(ui_section=section, ui_key=key).scalar()
     @classmethod
     def get_or_create(cls, section, key):
         """ Return specified Ui object, creating it if necessary. """
         setting = cls.get_by_key(section, key)
         if setting is None:
             setting = cls(ui_section=section, ui_key=key)
             Session().add(setting)
         return setting
     @classmethod
     def get_builtin_hooks(cls):
         q = cls.query()
         q = q.filter(cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE,
                                      cls.HOOK_PUSH, cls.HOOK_PRE_PUSH,
                                      cls.HOOK_PULL, cls.HOOK_PRE_PULL]))
         q = q.filter(cls.ui_section == 'hooks')
         return q.all()
     @classmethod
     def get_custom_hooks(cls):
         q = cls.query()
         q = q.filter(~cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE,
                                       cls.HOOK_PUSH, cls.HOOK_PRE_PUSH,
                                       cls.HOOK_PULL, cls.HOOK_PRE_PULL]))
         q = q.filter(cls.ui_section == 'hooks')
         return q.all()
     @classmethod
     def get_repos_location(cls):
         return cls.get_by_key('paths', '/').ui_value
     @classmethod
     def create_or_update_hook(cls, key, val):
         new_ui = cls.get_or_create('hooks', key)
         new_ui.ui_active = True
         new_ui.ui_value = val
     def __repr__(self):
         return '<%s[%s]%s=>%s]>' % (self.__class__.__name__, self.ui_section,
                                     self.ui_key, self.ui_value)
 class User(Base, BaseDbModel):
     __tablename__ = 'users'
     __table_args__ = (
         Index('u_username_idx', 'username'),
         Index('u_email_idx', 'email'),
         _table_args_default_dict,
+    )
     DEFAULT_USER = 'default'
     DEFAULT_GRAVATAR_URL = 'https://secure.gravatar.com/avatar/{md5email}?d=identicon&s={size}'
     # The name of the default auth type in extern_type, 'internal' lives in auth_internal.py
     DEFAULT_AUTH_TYPE = 'internal'
     user_id = Column(Integer(), primary_key=True)
     username = Column(String(255), nullable=False, unique=True)
     password = Column(String(255), nullable=False)
     active = Column(Boolean(), nullable=False, default=True)
     admin = Column(Boolean(), nullable=False, default=False)
     name = Column("firstname", Unicode(255), nullable=False)
     lastname = Column(Unicode(255), nullable=False)
     _email = Column("email", String(255), nullable=True, unique=True) # FIXME: not nullable?
     last_login = Column(DateTime(timezone=False), nullable=True)
     extern_type = Column(String(255), nullable=True) # FIXME: not nullable?
     extern_name = Column(String(255), nullable=True) # FIXME: not nullable?
     api_key = Column(String(255), nullable=False)
     inherit_default_permissions = Column(Boolean(), nullable=False, default=True)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     _user_data = Column("user_data", LargeBinary(), nullable=True)  # JSON data # FIXME: not nullable?
     user_log = relationship('UserLog')
     user_perms = relationship('UserToPerm', primaryjoin="User.user_id==UserToPerm.user_id", cascade='all')
     repositories = relationship('Repository')
     repo_groups = relationship('RepoGroup')
     user_groups = relationship('UserGroup')
     user_followers = relationship('UserFollowing', primaryjoin='UserFollowing.follows_user_id==User.user_id', cascade='all')
     followings = relationship('UserFollowing', primaryjoin='UserFollowing.user_id==User.user_id', cascade='all')
     repo_to_perm = relationship('UserRepoToPerm', primaryjoin='UserRepoToPerm.user_id==User.user_id', cascade='all')
     repo_group_to_perm = relationship('UserRepoGroupToPerm', primaryjoin='UserRepoGroupToPerm.user_id==User.user_id', cascade='all')
     group_member = relationship('UserGroupMember', cascade='all')
     notifications = relationship('UserNotification', cascade='all')
     # notifications assigned to this user
     user_created_notifications = relationship('Notification', cascade='all')
     # comments created by this user
     user_comments = relationship('ChangesetComment', cascade='all')
     #extra emails for this user
     user_emails = relationship('UserEmailMap', cascade='all')
     #extra API keys
     user_api_keys = relationship('UserApiKeys', cascade='all')
     @hybrid_property
     def email(self):
         return self._email
     @email.setter
     def email(self, val):
         self._email = val.lower() if val else None
     @property
     def firstname(self):
         # alias for future
         return self.name
     @property
     def emails(self):
         other = UserEmailMap.query().filter(UserEmailMap.user==self).all()
         return [self.email] + [x.email for x in other]
     @property
     def api_keys(self):
         other = UserApiKeys.query().filter(UserApiKeys.user==self).all()
         return [self.api_key] + [x.api_key for x in other]
     @property
     def ip_addresses(self):
         ret = UserIpMap.query().filter(UserIpMap.user == self).all()
         return [x.ip_addr for x in ret]
     @property
     def full_name(self):
         return '%s %s' % (self.firstname, self.lastname)
     @property
     def full_name_or_username(self):
         """
         Show full name.
         If full name is not set, fall back to username.
         """
         return ('%s %s' % (self.firstname, self.lastname)
                 if (self.firstname and self.lastname) else self.username)
     @property
     def full_name_and_username(self):
         """
         Show full name and username as 'Firstname Lastname (username)'.
         If full name is not set, fall back to username.
         """
         return ('%s %s (%s)' % (self.firstname, self.lastname, self.username)
                 if (self.firstname and self.lastname) else self.username)
     @property
     def full_contact(self):
         return '%s %s <%s>' % (self.firstname, self.lastname, self.email)
     @property
     def short_contact(self):
         return '%s %s' % (self.firstname, self.lastname)
     @property
     def is_admin(self):
         return self.admin
     @property
     def AuthUser(self):
         """
         Returns instance of AuthUser for this user
         """
         from kallithea.lib.auth import AuthUser
         return AuthUser(dbuser=self)
     @hybrid_property
     def user_data(self):
         if not self._user_data:
             return {}
         try:
             return json.loads(self._user_data)
         except TypeError:
             return {}
     @user_data.setter
     def user_data(self, val):
         try:
             self._user_data = json.dumps(val)
         except Exception:
             log.error(traceback.format_exc())
     def __unicode__(self):
         return u"<%s('id:%s:%s')>" % (self.__class__.__name__,
                                       self.user_id, self.username)
     @classmethod
     def guess_instance(cls, value):
         return super(User, cls).guess_instance(value, User.get_by_username)
     @classmethod
     def get_or_404(cls, id_, allow_default=True):
         '''
         Overridden version of BaseDbModel.get_or_404, with an extra check on
         the default user.
         '''
         user = super(User, cls).get_or_404(id_)
         if allow_default == False:
             if user.username == User.DEFAULT_USER:
                 raise DefaultUserException
         return user
     @classmethod
     def get_by_username_or_email(cls, username_or_email, case_insensitive=False, cache=False):
         """
         For anything that looks like an email address, look up by the email address (matching
         case insensitively).
         For anything else, try to look up by the user name.
         This assumes no normal username can have '@' symbol.
         """
         if '@' in username_or_email:
             return User.get_by_email(username_or_email, cache=cache)
         else:
             return User.get_by_username(username_or_email, case_insensitive=case_insensitive, cache=cache)
     @classmethod
     def get_by_username(cls, username, case_insensitive=False, cache=False):
         if case_insensitive:
             q = cls.query().filter(func.lower(cls.username) == func.lower(username))
         else:
             q = cls.query().filter(cls.username == username)
         if cache:
             q = q.options(FromCache(
                             "sql_cache_short",
                             "get_user_%s" % _hash_key(username)
+                          )
+            )
         return q.scalar()
     @classmethod
     def get_by_api_key(cls, api_key, cache=False, fallback=True):
         if len(api_key) != 40 or not api_key.isalnum():
             return None
         q = cls.query().filter(cls.api_key == api_key)
         if cache:
             q = q.options(FromCache("sql_cache_short",
                                     "get_api_key_%s" % api_key))
         res = q.scalar()
         if fallback and not res:
             #fallback to additional keys
             _res = UserApiKeys.query() \
                 .filter(UserApiKeys.api_key == api_key) \
                 .filter(or_(UserApiKeys.expires == -1,
                             UserApiKeys.expires >= time.time())) \
                 .first()
             if _res:
                 res = _res.user
         return res
     @classmethod
     def get_by_email(cls, email, cache=False):
         q = cls.query().filter(func.lower(cls.email) == func.lower(email))
         if cache:
             q = q.options(FromCache("sql_cache_short",
                                     "get_email_key_%s" % email))
         ret = q.scalar()
         if ret is None:
             q = UserEmailMap.query()
             # try fetching in alternate email map
             q = q.filter(func.lower(UserEmailMap.email) == func.lower(email))
             q = q.options(joinedload(UserEmailMap.user))
             if cache:
                 q = q.options(FromCache("sql_cache_short",
                                         "get_email_map_key_%s" % email))
             ret = getattr(q.scalar(), 'user', None)
         return ret
     @classmethod
     def get_from_cs_author(cls, author):
         """
         Tries to get User objects out of commit author string
         :param author:
         """
         from kallithea.lib.helpers import email, author_name
         # Valid email in the attribute passed, see if they're in the system
         _email = email(author)
         if _email:
             user = cls.get_by_email(_email)
             if user is not None:
                 return user
         # Maybe we can match by username?
         _author = author_name(author)
         user = cls.get_by_username(_author, case_insensitive=True)
         if user is not None:
             return user
     def update_lastlogin(self):
         """Update user lastlogin"""
         self.last_login = datetime.datetime.now()
         log.debug('updated user %s lastlogin', self.username)
     @classmethod
     def get_first_admin(cls):
         user = User.query().filter(User.admin == True).first()
         if user is None:
             raise Exception('Missing administrative account!')
         return user
     @classmethod
     def get_default_user(cls, cache=False):
         user = User.get_by_username(User.DEFAULT_USER, cache=cache)
         if user is None:
             raise Exception('Missing default account!')
         return user
     def get_api_data(self, details=False):
         """
         Common function for generating user related data for API
         """
         user = self
         data = dict(
             user_id=user.user_id,
             username=user.username,
             firstname=user.name,
             lastname=user.lastname,
             email=user.email,
             emails=user.emails,
             active=user.active,
             admin=user.admin,
+        )
         if details:
             data.update(dict(
                 extern_type=user.extern_type,
                 extern_name=user.extern_name,
                 api_key=user.api_key,
                 api_keys=user.api_keys,
                 last_login=user.last_login,
                 ip_addresses=user.ip_addresses
                 ))
         return data
     def __json__(self):
         data = dict(
             full_name=self.full_name,
             full_name_or_username=self.full_name_or_username,
             short_contact=self.short_contact,
             full_contact=self.full_contact
+        )
         data.update(self.get_api_data())
         return data
 class UserApiKeys(Base, BaseDbModel):
     __tablename__ = 'user_api_keys'
     __table_args__ = (
         Index('uak_api_key_idx', 'api_key'),
         Index('uak_api_key_expires_idx', 'api_key', 'expires'),
         _table_args_default_dict,
+    )
     __mapper_args__ = {}
     user_api_key_id = Column(Integer(), primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     api_key = Column(String(255), nullable=False, unique=True)
     description = Column(UnicodeText(), nullable=False)
     expires = Column(Float(53), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     user = relationship('User')
     @property
     def expired(self):
         if self.expires == -1:
             return False
         return time.time() > self.expires
 class UserEmailMap(Base, BaseDbModel):
     __tablename__ = 'user_email_map'
     __table_args__ = (
         Index('uem_email_idx', 'email'),
         _table_args_default_dict,
+    )
     __mapper_args__ = {}
     email_id = Column(Integer(), primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     _email = Column("email", String(255), nullable=False, unique=True)
     user = relationship('User')
     @validates('_email')
     def validate_email(self, key, email):
         # check if this email is not main one
         main_email = Session().query(User).filter(User.email == email).scalar()
         if main_email is not None:
             raise AttributeError('email %s is present is user table' % email)
         return email
     @hybrid_property
     def email(self):
         return self._email
     @email.setter
     def email(self, val):
         self._email = val.lower() if val else None
 class UserIpMap(Base, BaseDbModel):
     __tablename__ = 'user_ip_map'
     __table_args__ = (
         UniqueConstraint('user_id', 'ip_addr'),
         _table_args_default_dict,
+    )
     __mapper_args__ = {}
     ip_id = Column(Integer(), primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     ip_addr = Column(String(255), nullable=False)
     active = Column(Boolean(), nullable=False, default=True)
     user = relationship('User')
     @classmethod
     def _get_ip_range(cls, ip_addr):
         from kallithea.lib import ipaddr
         net = ipaddr.IPNetwork(address=ip_addr)
         return [str(net.network), str(net.broadcast)]
     def __json__(self):
         return dict(
           ip_addr=self.ip_addr,
           ip_range=self._get_ip_range(self.ip_addr)
+        )
     def __unicode__(self):
         return u"<%s('user_id:%s=>%s')>" % (self.__class__.__name__,
                                             self.user_id, self.ip_addr)
 class UserLog(Base, BaseDbModel):
     __tablename__ = 'user_logs'
     __table_args__ = (
         _table_args_default_dict,
+    )
     user_log_id = Column(Integer(), primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=True)
     username = Column(String(255), nullable=False)
     repository_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=True)
     repository_name = Column(Unicode(255), nullable=False)
     user_ip = Column(String(255), nullable=True)
     action = Column(UnicodeText(), nullable=False)
     action_date = Column(DateTime(timezone=False), nullable=False)
     def __unicode__(self):
         return u"<%s('id:%s:%s')>" % (self.__class__.__name__,
                                       self.repository_name,
                                       self.action)
     @property
     def action_as_day(self):
         return datetime.date(*self.action_date.timetuple()[:3])
     user = relationship('User')
     repository = relationship('Repository', cascade='')
 class UserGroup(Base, BaseDbModel):
     __tablename__ = 'users_groups'
     __table_args__ = (
         _table_args_default_dict,
+    )
     users_group_id = Column(Integer(), primary_key=True)
     users_group_name = Column(Unicode(255), nullable=False, unique=True)
     user_group_description = Column(Unicode(10000), nullable=True) # FIXME: not nullable?
     users_group_active = Column(Boolean(), nullable=False)
     inherit_default_permissions = Column("users_group_inherit_default_permissions", Boolean(), nullable=False, default=True)
     owner_id = Column('user_id', Integer(), ForeignKey('users.user_id'), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     _group_data = Column("group_data", LargeBinary(), nullable=True)  # JSON data # FIXME: not nullable?
     members = relationship('UserGroupMember', cascade="all, delete-orphan")
     users_group_to_perm = relationship('UserGroupToPerm', cascade='all')
     users_group_repo_to_perm = relationship('UserGroupRepoToPerm', cascade='all')
     users_group_repo_group_to_perm = relationship('UserGroupRepoGroupToPerm', cascade='all')
     user_user_group_to_perm = relationship('UserUserGroupToPerm ', cascade='all')
     user_group_user_group_to_perm = relationship('UserGroupUserGroupToPerm ', primaryjoin="UserGroupUserGroupToPerm.target_user_group_id==UserGroup.users_group_id", cascade='all')
     owner = relationship('User')
     @hybrid_property
     def group_data(self):
         if not self._group_data:
             return {}
         try:
             return json.loads(self._group_data)
         except TypeError:
             return {}
     @group_data.setter
     def group_data(self, val):
         try:
             self._group_data = json.dumps(val)
         except Exception:
             log.error(traceback.format_exc())
     def __unicode__(self):
         return u"<%s('id:%s:%s')>" % (self.__class__.__name__,
                                       self.users_group_id,
                                       self.users_group_name)
     @classmethod
     def guess_instance(cls, value):
         return super(UserGroup, cls).guess_instance(value, UserGroup.get_by_group_name)
     @classmethod
     def get_by_group_name(cls, group_name, cache=False,
                           case_insensitive=False):
         if case_insensitive:
             q = cls.query().filter(func.lower(cls.users_group_name) == func.lower(group_name))
         else:
             q = cls.query().filter(cls.users_group_name == group_name)
         if cache:
             q = q.options(FromCache(
                             "sql_cache_short",
                             "get_group_%s" % _hash_key(group_name)
+                          )
+            )
         return q.scalar()
     @classmethod
     def get(cls, user_group_id, cache=False):
         user_group = cls.query()
         if cache:
             user_group = user_group.options(FromCache("sql_cache_short",
                                     "get_users_group_%s" % user_group_id))
         return user_group.get(user_group_id)
     def get_api_data(self, with_members=True):
         user_group = self
         data = dict(
             users_group_id=user_group.users_group_id,
             group_name=user_group.users_group_name,
             group_description=user_group.user_group_description,
             active=user_group.users_group_active,
             owner=user_group.owner.username,
+        )
         if with_members:
             members = []
             for user in user_group.members:
                 user = user.user
                 members.append(user.get_api_data())
             data['members'] = members
         return data
 class UserGroupMember(Base, BaseDbModel):
     __tablename__ = 'users_groups_members'
     __table_args__ = (
         _table_args_default_dict,
+    )
     users_group_member_id = Column(Integer(), primary_key=True)
     users_group_id = Column(Integer(), ForeignKey('users_groups.users_group_id'), nullable=False)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     user = relationship('User')
     users_group = relationship('UserGroup')
     def __init__(self, gr_id='', u_id=''):
         self.users_group_id = gr_id
         self.user_id = u_id
 class RepositoryField(Base, BaseDbModel):
     __tablename__ = 'repositories_fields'
     __table_args__ = (
         UniqueConstraint('repository_id', 'field_key'),  # no-multi field
         _table_args_default_dict,
+    )
     PREFIX = 'ex_'  # prefix used in form to not conflict with already existing fields
     repo_field_id = Column(Integer(), primary_key=True)
     repository_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)
     field_key = Column(String(250), nullable=False)
     field_label = Column(String(1024), nullable=False)
     field_value = Column(String(10000), nullable=False)
     field_desc = Column(String(1024), nullable=False)
     field_type = Column(String(255), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     repository = relationship('Repository')
     @property
     def field_key_prefixed(self):
         return 'ex_%s' % self.field_key
     @classmethod
     def un_prefix_key(cls, key):
         if key.startswith(cls.PREFIX):
             return key[len(cls.PREFIX):]
         return key
     @classmethod
     def get_by_key_name(cls, key, repo):
         row = cls.query() \
                 .filter(cls.repository == repo) \
                 .filter(cls.field_key == key).scalar()
         return row
 class Repository(Base, BaseDbModel):
     __tablename__ = 'repositories'
     __table_args__ = (
         Index('r_repo_name_idx', 'repo_name'),
         _table_args_default_dict,
+    )
     DEFAULT_CLONE_URI = '{scheme}://{user}@{netloc}/{repo}'
     DEFAULT_CLONE_URI_ID = '{scheme}://{user}@{netloc}/_{repoid}'
     STATE_CREATED = u'repo_state_created'
     STATE_PENDING = u'repo_state_pending'
     STATE_ERROR = u'repo_state_error'
     repo_id = Column(Integer(), primary_key=True)
     repo_name = Column(Unicode(255), nullable=False, unique=True)
     repo_state = Column(String(255), nullable=False)
     clone_uri = Column(String(255), nullable=True) # FIXME: not nullable?
     repo_type = Column(String(255), nullable=False)
     owner_id = Column('user_id', Integer(), ForeignKey('users.user_id'), nullable=False)
     private = Column(Boolean(), nullable=False)
     enable_statistics = Column("statistics", Boolean(), nullable=False, default=True)
     enable_downloads = Column("downloads", Boolean(), nullable=False, default=True)
     description = Column(Unicode(10000), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     updated_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     _landing_revision = Column("landing_revision", String(255), nullable=False)
     enable_locking = Column(Boolean(), nullable=False, default=False)
     _locked = Column("locked", String(255), nullable=True) # FIXME: not nullable?
     _changeset_cache = Column("changeset_cache", LargeBinary(), nullable=True) # JSON data # FIXME: not nullable?
     fork_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=True)
     group_id = Column(Integer(), ForeignKey('groups.group_id'), nullable=True)
     owner = relationship('User')
     fork = relationship('Repository', remote_side=repo_id)
     group = relationship('RepoGroup')
     repo_to_perm = relationship('UserRepoToPerm', cascade='all', order_by='UserRepoToPerm.repo_to_perm_id')
     users_group_to_perm = relationship('UserGroupRepoToPerm', cascade='all')
     stats = relationship('Statistics', cascade='all', uselist=False)
     followers = relationship('UserFollowing',
                              primaryjoin='UserFollowing.follows_repository_id==Repository.repo_id',
                              cascade='all')
     extra_fields = relationship('RepositoryField',
                                 cascade="all, delete-orphan")
     logs = relationship('UserLog')
     comments = relationship('ChangesetComment', cascade="all, delete-orphan")
     pull_requests_org = relationship('PullRequest',
                     primaryjoin='PullRequest.org_repo_id==Repository.repo_id',
                     cascade="all, delete-orphan")
     pull_requests_other = relationship('PullRequest',
                     primaryjoin='PullRequest.other_repo_id==Repository.repo_id',
                     cascade="all, delete-orphan")
     def __unicode__(self):
         return u"<%s('%s:%s')>" % (self.__class__.__name__, self.repo_id,
                                    safe_unicode(self.repo_name))
     @hybrid_property
     def landing_rev(self):
         # always should return [rev_type, rev]
         if self._landing_revision:
             _rev_info = self._landing_revision.split(':')
             if len(_rev_info) < 2:
                 _rev_info.insert(0, 'rev')
             return [_rev_info[0], _rev_info[1]]
         return [None, None]
     @landing_rev.setter
     def landing_rev(self, val):
         if ':' not in val:
             raise ValueError('value must be delimited with `:` and consist '
                              'of <rev_type>:<rev>, got %s instead' % val)
         self._landing_revision = val
     @hybrid_property
     def locked(self):
         # always should return [user_id, timelocked]
         if self._locked:
             _lock_info = self._locked.split(':')
             return int(_lock_info[0]), _lock_info[1]
         return [None, None]
     @locked.setter
     def locked(self, val):
         if val and isinstance(val, (list, tuple)):
             self._locked = ':'.join(map(str, val))
         else:
             self._locked = None
     @hybrid_property
     def changeset_cache(self):
         try:
             cs_cache = json.loads(self._changeset_cache) # might raise on bad data
             cs_cache['raw_id'] # verify data, raise exception on error
             return cs_cache
         except (TypeError, KeyError, ValueError):
             return EmptyChangeset().__json__()
     @changeset_cache.setter
     def changeset_cache(self, val):
         try:
             self._changeset_cache = json.dumps(val)
         except Exception:
             log.error(traceback.format_exc())
     @classmethod
     def query(cls, sorted=False):
         """Add Repository-specific helpers for common query constructs.
         sorted: if True, apply the default ordering (name, case insensitive).
         """
         q = super(Repository, cls).query()
         if sorted:
             q = q.order_by(func.lower(Repository.repo_name))
         return q
     @classmethod
     def url_sep(cls):
         return URL_SEP
     @classmethod
     def normalize_repo_name(cls, repo_name):
         """
         Normalizes os specific repo_name to the format internally stored inside
         database using URL_SEP
         :param cls:
         :param repo_name:
         """
         return cls.url_sep().join(repo_name.split(os.sep))
     @classmethod
     def guess_instance(cls, value):
         return super(Repository, cls).guess_instance(value, Repository.get_by_repo_name)
     @classmethod
     def get_by_repo_name(cls, repo_name):
         q = Session().query(cls).filter(cls.repo_name == repo_name)
         q = q.options(joinedload(Repository.fork)) \
                 .options(joinedload(Repository.owner)) \
                 .options(joinedload(Repository.group))
         return q.scalar()
     @classmethod
     def get_by_full_path(cls, repo_full_path):
         base_full_path = os.path.realpath(cls.base_path())
         repo_full_path = os.path.realpath(repo_full_path)
         assert repo_full_path.startswith(base_full_path + os.path.sep)
         repo_name = repo_full_path[len(base_full_path) + 1:]
         repo_name = cls.normalize_repo_name(repo_name)
         return cls.get_by_repo_name(repo_name.strip(URL_SEP))
     @classmethod
     def get_repo_forks(cls, repo_id):
         return cls.query().filter(Repository.fork_id == repo_id)
     @classmethod
     def base_path(cls):
         """
         Returns base path where all repos are stored
         :param cls:
         """
         q = Session().query(Ui) \
             .filter(Ui.ui_key == cls.url_sep())
         q = q.options(FromCache("sql_cache_short", "repository_repo_path"))
         return q.one().ui_value
     @property
     def forks(self):
         """
         Return forks of this repo
         """
         return Repository.get_repo_forks(self.repo_id)
     @property
     def parent(self):
         """
         Returns fork parent
         """
         return self.fork
     @property
     def just_name(self):
         return self.repo_name.split(Repository.url_sep())[-1]
     @property
     def groups_with_parents(self):
         groups = []
         group = self.group
         while group is not None:
             groups.append(group)
             group = group.parent_group
             assert group not in groups, group # avoid recursion on bad db content
         groups.reverse()
         return groups
     @LazyProperty
     def repo_path(self):
         """
         Returns base full path for that repository means where it actually
         exists on a filesystem
         """
         q = Session().query(Ui).filter(Ui.ui_key ==
                                               Repository.url_sep())
         q = q.options(FromCache("sql_cache_short", "repository_repo_path"))
         return q.one().ui_value
     @property
     def repo_full_path(self):
         p = [self.repo_path]
         # we need to split the name by / since this is how we store the
         # names in the database, but that eventually needs to be converted
         # into a valid system path
         p += self.repo_name.split(Repository.url_sep())
         return os.path.join(*map(safe_unicode, p))
     @property
     def cache_keys(self):
         """
         Returns associated cache keys for that repo
         """
         return CacheInvalidation.query() \
             .filter(CacheInvalidation.cache_args == self.repo_name) \
             .order_by(CacheInvalidation.cache_key) \
             .all()
     def get_new_name(self, repo_name):
         """
         returns new full repository name based on assigned group and new new
         :param group_name:
         """
         path_prefix = self.group.full_path_splitted if self.group else []
         return Repository.url_sep().join(path_prefix + [repo_name])
     @property
     def _ui(self):
         """
         Creates an db based ui object for this repository
         """
         from kallithea.lib.utils import make_ui
         return make_ui('db', clear_session=False)
     @classmethod
     def is_valid(cls, repo_name):
         """
         returns True if given repo name is a valid filesystem repository
         :param cls:
         :param repo_name:
         """
         from kallithea.lib.utils import is_valid_repo
         return is_valid_repo(repo_name, cls.base_path())
     def get_api_data(self):
         """
         Common function for generating repo api data
         """
         repo = self
         data = dict(
             repo_id=repo.repo_id,
             repo_name=repo.repo_name,
             repo_type=repo.repo_type,
             clone_uri=repo.clone_uri,
             private=repo.private,
             created_on=repo.created_on,
             description=repo.description,
             landing_rev=repo.landing_rev,
             owner=repo.owner.username,
             fork_of=repo.fork.repo_name if repo.fork else None,
             enable_statistics=repo.enable_statistics,
             enable_locking=repo.enable_locking,
             enable_downloads=repo.enable_downloads,
             last_changeset=repo.changeset_cache,
             locked_by=User.get(self.locked[0]).get_api_data() \
                 if self.locked[0] else None,
             locked_date=time_to_datetime(self.locked[1]) \
                 if self.locked[1] else None
+        )
         rc_config = Setting.get_app_settings()
         repository_fields = str2bool(rc_config.get('repository_fields'))
         if repository_fields:
             for f in self.extra_fields:
                 data[f.field_key_prefixed] = f.field_value
         return data
     @classmethod
     def lock(cls, repo, user_id, lock_time=None):
         if lock_time is not None:
             lock_time = time.time()
         repo.locked = [user_id, lock_time]
         Session().commit()
     @classmethod
     def unlock(cls, repo):
         repo.locked = None
         Session().commit()
     @classmethod
     def getlock(cls, repo):
         return repo.locked
     @property
     def last_db_change(self):
         return self.updated_on
     @property
     def clone_uri_hidden(self):
         clone_uri = self.clone_uri
         if clone_uri:
             import urlobject
             url_obj = urlobject.URLObject(self.clone_uri)
             if url_obj.password:
                 clone_uri = url_obj.with_password('*****')
         return clone_uri
     def clone_url(self, **override):
         import kallithea.lib.helpers as h
         qualified_home_url = h.canonical_url('home')
         uri_tmpl = None
         if 'with_id' in override:
             uri_tmpl = self.DEFAULT_CLONE_URI_ID
             del override['with_id']
@@ @@ -1373,600 +1385,608 @@ class Repository(Base, BaseDbModel): @@
                 cs_cache = scm_repo.get_changeset()
         if isinstance(cs_cache, BaseChangeset):
             cs_cache = cs_cache.__json__()
         if (not self.changeset_cache or cs_cache['raw_id'] != self.changeset_cache['raw_id']):
             _default = datetime.datetime.fromtimestamp(0)
             last_change = cs_cache.get('date') or _default
             log.debug('updated repo %s with new cs cache %s',
                       self.repo_name, cs_cache)
             self.updated_on = last_change
             self.changeset_cache = cs_cache
             Session().commit()
         else:
             log.debug('changeset_cache for %s already up to date with %s',
                       self.repo_name, cs_cache['raw_id'])
     @property
     def tip(self):
         return self.get_changeset('tip')
     @property
     def author(self):
         return self.tip.author
     @property
     def last_change(self):
         return self.scm_instance.last_change
     def get_comments(self, revisions=None):
         """
         Returns comments for this repository grouped by revisions
         :param revisions: filter query by revisions only
         """
         cmts = ChangesetComment.query() \
             .filter(ChangesetComment.repo == self)
         if revisions is not None:
             if not revisions:
                 return {} # don't use sql 'in' on empty set
             cmts = cmts.filter(ChangesetComment.revision.in_(revisions))
         grouped = collections.defaultdict(list)
         for cmt in cmts.all():
             grouped[cmt.revision].append(cmt)
         return grouped
     def statuses(self, revisions):
         """
         Returns statuses for this repository.
         PRs without any votes do _not_ show up as unreviewed.
         :param revisions: list of revisions to get statuses for
         """
         if not revisions:
             return {}
         statuses = ChangesetStatus.query() \
             .filter(ChangesetStatus.repo == self) \
             .filter(ChangesetStatus.version == 0) \
             .filter(ChangesetStatus.revision.in_(revisions))
         grouped = {}
         for stat in statuses.all():
             pr_id = pr_nice_id = pr_repo = None
             if stat.pull_request:
                 pr_id = stat.pull_request.pull_request_id
                 pr_nice_id = PullRequest.make_nice_id(pr_id)
                 pr_repo = stat.pull_request.other_repo.repo_name
             grouped[stat.revision] = [str(stat.status), stat.status_lbl,
                                       pr_id, pr_repo, pr_nice_id,
                                       stat.author]
         return grouped
     def _repo_size(self):
         from kallithea.lib import helpers as h
         log.debug('calculating repository size...')
         return h.format_byte_size(self.scm_instance.size)
     #==========================================================================
     # SCM CACHE INSTANCE
     #==========================================================================
     def set_invalidate(self):
         """
         Mark caches of this repo as invalid.
         """
         CacheInvalidation.set_invalidate(self.repo_name)
     _scm_instance = None
     @property
     def scm_instance(self):
         if self._scm_instance is None:
             self._scm_instance = self.scm_instance_cached()
         return self._scm_instance
     def scm_instance_cached(self, valid_cache_keys=None):
         @cache_region('long_term', 'scm_instance_cached')
         def _c(repo_name): # repo_name is just for the cache key
             log.debug('Creating new %s scm_instance and populating cache', repo_name)
             return self.scm_instance_no_cache()
         rn = self.repo_name
         valid = CacheInvalidation.test_and_set_valid(rn, None, valid_cache_keys=valid_cache_keys)
         if not valid:
             log.debug('Cache for %s invalidated, getting new object', rn)
             region_invalidate(_c, None, 'scm_instance_cached', rn)
         else:
             log.debug('Trying to get scm_instance of %s from cache', rn)
         return _c(rn)
     def scm_instance_no_cache(self):
         repo_full_path = safe_str(self.repo_full_path)
         alias = get_scm(repo_full_path)[0]
         log.debug('Creating instance of %s repository from %s',
                   alias, self.repo_full_path)
         backend = get_backend(alias)
         if alias == 'hg':
             repo = backend(repo_full_path, create=False,
                            baseui=self._ui)
         else:
             repo = backend(repo_full_path, create=False)
         return repo
     def __json__(self):
         return dict(landing_rev = self.landing_rev)
 class RepoGroup(Base, BaseDbModel):
     __tablename__ = 'groups'
     __table_args__ = (
         CheckConstraint('group_id != group_parent_id', name='ck_groups_no_self_parent'),
         _table_args_default_dict,
+    )
     __mapper_args__ = {'order_by': 'group_name'} # TODO: Deprecated as of SQLAlchemy 1.1.
     SEP = ' &raquo; '
     group_id = Column(Integer(), primary_key=True)
     group_name = Column(Unicode(255), nullable=False, unique=True) # full path
     parent_group_id = Column('group_parent_id', Integer(), ForeignKey('groups.group_id'), nullable=True)
     group_description = Column(Unicode(10000), nullable=False)
     enable_locking = Column(Boolean(), nullable=False, default=False)
     owner_id = Column('user_id', Integer(), ForeignKey('users.user_id'), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     repo_group_to_perm = relationship('UserRepoGroupToPerm', cascade='all', order_by='UserRepoGroupToPerm.group_to_perm_id')
     users_group_to_perm = relationship('UserGroupRepoGroupToPerm', cascade='all')
     parent_group = relationship('RepoGroup', remote_side=group_id)
     owner = relationship('User')
     @classmethod
     def query(cls, sorted=False):
         """Add RepoGroup-specific helpers for common query constructs.
         sorted: if True, apply the default ordering (name, case insensitive).
         """
         q = super(RepoGroup, cls).query()
         if sorted:
             q = q.order_by(func.lower(RepoGroup.group_name))
         return q
     def __init__(self, group_name='', parent_group=None):
         self.group_name = group_name
         self.parent_group = parent_group
     def __unicode__(self):
         return u"<%s('id:%s:%s')>" % (self.__class__.__name__, self.group_id,
                                       self.group_name)
     @classmethod
     def _generate_choice(cls, repo_group):
         """Return tuple with group_id and name as html literal"""
         from webhelpers.html import literal
         if repo_group is None:
             return (-1, u'-- %s --' % _('top level'))
         return repo_group.group_id, literal(cls.SEP.join(repo_group.full_path_splitted))
     @classmethod
     def groups_choices(cls, groups):
         """Return tuples with group_id and name as html literal."""
         return sorted((cls._generate_choice(g) for g in groups),
                       key=lambda c: c[1].split(cls.SEP))
     @classmethod
     def url_sep(cls):
         return URL_SEP
     @classmethod
     def guess_instance(cls, value):
         return super(RepoGroup, cls).guess_instance(value, RepoGroup.get_by_group_name)
     @classmethod
     def get_by_group_name(cls, group_name, cache=False, case_insensitive=False):
         group_name = group_name.rstrip('/')
         if case_insensitive:
             gr = cls.query() \
                 .filter(func.lower(cls.group_name) == func.lower(group_name))
         else:
             gr = cls.query() \
                 .filter(cls.group_name == group_name)
         if cache:
             gr = gr.options(FromCache(
                             "sql_cache_short",
                             "get_group_%s" % _hash_key(group_name)
+                            )
+            )
         return gr.scalar()
     @property
     def parents(self):
         groups = []
         group = self.parent_group
         while group is not None:
             groups.append(group)
             group = group.parent_group
             assert group not in groups, group # avoid recursion on bad db content
         groups.reverse()
         return groups
     @property
     def children(self):
         return RepoGroup.query().filter(RepoGroup.parent_group == self)
     @property
     def name(self):
         return self.group_name.split(RepoGroup.url_sep())[-1]
     @property
     def full_path(self):
         return self.group_name
     @property
     def full_path_splitted(self):
         return self.group_name.split(RepoGroup.url_sep())
     @property
     def repositories(self):
         return Repository.query(sorted=True).filter_by(group=self)
     @property
     def repositories_recursive_count(self):
         cnt = self.repositories.count()
         def children_count(group):
             cnt = 0
             for child in group.children:
                 cnt += child.repositories.count()
                 cnt += children_count(child)
             return cnt
         return cnt + children_count(self)
     def _recursive_objects(self, include_repos=True):
         all_ = []
         def _get_members(root_gr):
             if include_repos:
                 for r in root_gr.repositories:
                     all_.append(r)
             childs = root_gr.children.all()
             if childs:
                 for gr in childs:
                     all_.append(gr)
                     _get_members(gr)
         _get_members(self)
         return [self] + all_
     def recursive_groups_and_repos(self):
         """
         Recursive return all groups, with repositories in those groups
         """
         return self._recursive_objects()
     def recursive_groups(self):
         """
         Returns all children groups for this group including children of children
         """
         return self._recursive_objects(include_repos=False)
     def get_new_name(self, group_name):
         """
         returns new full group name based on parent and new name
         :param group_name:
         """
         path_prefix = (self.parent_group.full_path_splitted if
                        self.parent_group else [])
         return RepoGroup.url_sep().join(path_prefix + [group_name])
     def get_api_data(self):
         """
         Common function for generating api data
         """
         group = self
         data = dict(
             group_id=group.group_id,
             group_name=group.group_name,
             group_description=group.group_description,
             parent_group=group.parent_group.group_name if group.parent_group else None,
             repositories=[x.repo_name for x in group.repositories],
             owner=group.owner.username
+        )
         return data
 class Permission(Base, BaseDbModel):
     __tablename__ = 'permissions'
     __table_args__ = (
         Index('p_perm_name_idx', 'permission_name'),
         _table_args_default_dict,
+    )
     PERMS = [
         ('hg.admin', _('Kallithea Administrator')),
         ('repository.none', _('Default user has no access to new repositories')),
         ('repository.read', _('Default user has read access to new repositories')),
         ('repository.write', _('Default user has write access to new repositories')),
         ('repository.admin', _('Default user has admin access to new repositories')),
         ('group.none', _('Default user has no access to new repository groups')),
         ('group.read', _('Default user has read access to new repository groups')),
         ('group.write', _('Default user has write access to new repository groups')),
         ('group.admin', _('Default user has admin access to new repository groups')),
         ('usergroup.none', _('Default user has no access to new user groups')),
         ('usergroup.read', _('Default user has read access to new user groups')),
         ('usergroup.write', _('Default user has write access to new user groups')),
         ('usergroup.admin', _('Default user has admin access to new user groups')),
         ('hg.repogroup.create.false', _('Only admins can create repository groups')),
         ('hg.repogroup.create.true', _('Non-admins can create repository groups')),
         ('hg.usergroup.create.false', _('Only admins can create user groups')),
         ('hg.usergroup.create.true', _('Non-admins can create user groups')),
         ('hg.create.none', _('Only admins can create top level repositories')),
         ('hg.create.repository', _('Non-admins can create top level repositories')),
         ('hg.create.write_on_repogroup.true', _('Repository creation enabled with write permission to a repository group')),
         ('hg.create.write_on_repogroup.false', _('Repository creation disabled with write permission to a repository group')),
         ('hg.fork.none', _('Only admins can fork repositories')),
         ('hg.fork.repository', _('Non-admins can fork repositories')),
         ('hg.register.none', _('Registration disabled')),
         ('hg.register.manual_activate', _('User registration with manual account activation')),
         ('hg.register.auto_activate', _('User registration with automatic account activation')),
         ('hg.extern_activate.manual', _('Manual activation of external account')),
         ('hg.extern_activate.auto', _('Automatic activation of external account')),
+    ]
     #definition of system default permissions for DEFAULT user
     DEFAULT_USER_PERMISSIONS = [
         'repository.read',
         'group.read',
         'usergroup.read',
         'hg.create.repository',
         'hg.create.write_on_repogroup.true',
         'hg.fork.repository',
         'hg.register.manual_activate',
         'hg.extern_activate.auto',
+    ]
     # defines which permissions are more important higher the more important
     # Weight defines which permissions are more important.
     # The higher number the more important.
     PERM_WEIGHTS = {
         'repository.none': 0,
         'repository.read': 1,
         'repository.write': 3,
         'repository.admin': 4,
         'group.none': 0,
         'group.read': 1,
         'group.write': 3,
         'group.admin': 4,
         'usergroup.none': 0,
         'usergroup.read': 1,
         'usergroup.write': 3,
         'usergroup.admin': 4,
         'hg.repogroup.create.false': 0,
         'hg.repogroup.create.true': 1,
         'hg.usergroup.create.false': 0,
         'hg.usergroup.create.true': 1,
         'hg.fork.none': 0,
         'hg.fork.repository': 1,
         'hg.create.none': 0,
         'hg.create.repository': 1
+    }
     permission_id = Column(Integer(), primary_key=True)
     permission_name = Column(String(255), nullable=False)
     def __unicode__(self):
         return u"<%s('%s:%s')>" % (
             self.__class__.__name__, self.permission_id, self.permission_name
+        )
     @classmethod
     def guess_instance(cls, value):
         return super(Permission, cls).guess_instance(value, Permission.get_by_key)
     @classmethod
     def get_by_key(cls, key):
         return cls.query().filter(cls.permission_name == key).scalar()
     @classmethod
     def get_default_perms(cls, default_user_id):
         q = Session().query(UserRepoToPerm, Repository, cls) \
          .join((Repository, UserRepoToPerm.repository_id == Repository.repo_id)) \
          .join((cls, UserRepoToPerm.permission_id == cls.permission_id)) \
          .filter(UserRepoToPerm.user_id == default_user_id)
         return q.all()
     @classmethod
     def get_default_group_perms(cls, default_user_id):
         q = Session().query(UserRepoGroupToPerm, RepoGroup, cls) \
          .join((RepoGroup, UserRepoGroupToPerm.group_id == RepoGroup.group_id)) \
          .join((cls, UserRepoGroupToPerm.permission_id == cls.permission_id)) \
          .filter(UserRepoGroupToPerm.user_id == default_user_id)
         return q.all()
     @classmethod
     def get_default_user_group_perms(cls, default_user_id):
         q = Session().query(UserUserGroupToPerm, UserGroup, cls) \
          .join((UserGroup, UserUserGroupToPerm.user_group_id == UserGroup.users_group_id)) \
          .join((cls, UserUserGroupToPerm.permission_id == cls.permission_id)) \
          .filter(UserUserGroupToPerm.user_id == default_user_id)
         return q.all()
 class UserRepoToPerm(Base, BaseDbModel):
     __tablename__ = 'repo_to_perm'
     __table_args__ = (
         UniqueConstraint('user_id', 'repository_id', 'permission_id'),
         _table_args_default_dict,
+    )
     repo_to_perm_id = Column(Integer(), primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
     repository_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)
     user = relationship('User')
     repository = relationship('Repository')
     permission = relationship('Permission')
     @classmethod
     def create(cls, user, repository, permission):
         n = cls()
         n.user = user
         n.repository = repository
         n.permission = permission
         Session().add(n)
         return n
     def __unicode__(self):
         return u'<%s => %s >' % (self.user, self.repository)
 class UserUserGroupToPerm(Base, BaseDbModel):
     __tablename__ = 'user_user_group_to_perm'
     __table_args__ = (
         UniqueConstraint('user_id', 'user_group_id', 'permission_id'),
         _table_args_default_dict,
+    )
     user_user_group_to_perm_id = Column(Integer(), primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
     user_group_id = Column(Integer(), ForeignKey('users_groups.users_group_id'), nullable=False)
     user = relationship('User')
     user_group = relationship('UserGroup')
     permission = relationship('Permission')
     @classmethod
     def create(cls, user, user_group, permission):
         n = cls()
         n.user = user
         n.user_group = user_group
         n.permission = permission
         Session().add(n)
         return n
     def __unicode__(self):
         return u'<%s => %s >' % (self.user, self.user_group)
 class UserToPerm(Base, BaseDbModel):
     __tablename__ = 'user_to_perm'
     __table_args__ = (
         UniqueConstraint('user_id', 'permission_id'),
         _table_args_default_dict,
+    )
     user_to_perm_id = Column(Integer(), primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
     user = relationship('User')
     permission = relationship('Permission')
     def __unicode__(self):
         return u'<%s => %s >' % (self.user, self.permission)
 class UserGroupRepoToPerm(Base, BaseDbModel):
     __tablename__ = 'users_group_repo_to_perm'
     __table_args__ = (
         UniqueConstraint('repository_id', 'users_group_id', 'permission_id'),
         _table_args_default_dict,
+    )
     users_group_to_perm_id = Column(Integer(), primary_key=True)
     users_group_id = Column(Integer(), ForeignKey('users_groups.users_group_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
     repository_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)
     users_group = relationship('UserGroup')
     permission = relationship('Permission')
     repository = relationship('Repository')
     @classmethod
     def create(cls, users_group, repository, permission):
         n = cls()
         n.users_group = users_group
         n.repository = repository
         n.permission = permission
         Session().add(n)
         return n
     def __unicode__(self):
         return u'<UserGroupRepoToPerm:%s => %s >' % (self.users_group, self.repository)
 class UserGroupUserGroupToPerm(Base, BaseDbModel):
     __tablename__ = 'user_group_user_group_to_perm'
     __table_args__ = (
         UniqueConstraint('target_user_group_id', 'user_group_id', 'permission_id'),
         CheckConstraint('target_user_group_id != user_group_id', name='ck_user_group_user_group_to_perm_no_self_target'),
         _table_args_default_dict,
+    )
     user_group_user_group_to_perm_id = Column(Integer(), primary_key=True)
     target_user_group_id = Column(Integer(), ForeignKey('users_groups.users_group_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
     user_group_id = Column(Integer(), ForeignKey('users_groups.users_group_id'), nullable=False)
     target_user_group = relationship('UserGroup', primaryjoin='UserGroupUserGroupToPerm.target_user_group_id==UserGroup.users_group_id')
     user_group = relationship('UserGroup', primaryjoin='UserGroupUserGroupToPerm.user_group_id==UserGroup.users_group_id')
     permission = relationship('Permission')
     @classmethod
     def create(cls, target_user_group, user_group, permission):
         n = cls()
         n.target_user_group = target_user_group
         n.user_group = user_group
         n.permission = permission
         Session().add(n)
         return n
     def __unicode__(self):
         return u'<UserGroupUserGroup:%s => %s >' % (self.target_user_group, self.user_group)
 class UserGroupToPerm(Base, BaseDbModel):
     __tablename__ = 'users_group_to_perm'
     __table_args__ = (
         UniqueConstraint('users_group_id', 'permission_id',),
         _table_args_default_dict,
+    )
     users_group_to_perm_id = Column(Integer(), primary_key=True)
     users_group_id = Column(Integer(), ForeignKey('users_groups.users_group_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
     users_group = relationship('UserGroup')
     permission = relationship('Permission')
 class UserRepoGroupToPerm(Base, BaseDbModel):
     __tablename__ = 'user_repo_group_to_perm'
     __table_args__ = (
         UniqueConstraint('user_id', 'group_id', 'permission_id'),
         _table_args_default_dict,
+    )
     group_to_perm_id = Column(Integer(), primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     group_id = Column(Integer(), ForeignKey('groups.group_id'), nullable=False)
     permission_id = Column(Integer(), ForeignKey('permissions.permission_id'), nullable=False)
@@ @@ -2349,253 +2369,257 @@ class PullRequest(Base, BaseDbModel): @@
         sorted: if True, apply the default ordering (newest first).
         """
         q = super(PullRequest, cls).query()
         if reviewer_id is not None:
             q = q.join(PullRequestReviewer).filter(PullRequestReviewer.user_id == reviewer_id)
         if not include_closed:
             q = q.filter(PullRequest.status != PullRequest.STATUS_CLOSED)
         if sorted:
             q = q.order_by(PullRequest.created_on.desc())
         return q
     def get_reviewer_users(self):
         """Like .reviewers, but actually returning the users"""
         return User.query() \
             .join(PullRequestReviewer) \
             .filter(PullRequestReviewer.pull_request == self) \
             .order_by(PullRequestReviewer.pull_request_reviewers_id) \
             .all()
     def is_closed(self):
         return self.status == self.STATUS_CLOSED
     def user_review_status(self, user_id):
         """Return the user's latest status votes on PR"""
         # note: no filtering on repo - that would be redundant
         status = ChangesetStatus.query() \
             .filter(ChangesetStatus.pull_request == self) \
             .filter(ChangesetStatus.user_id == user_id) \
             .order_by(ChangesetStatus.version) \
             .first()
         return str(status.status) if status else ''
     @classmethod
     def make_nice_id(cls, pull_request_id):
         '''Return pull request id nicely formatted for displaying'''
         return '#%s' % pull_request_id
     def nice_id(self):
         '''Return the id of this pull request, nicely formatted for displaying'''
         return self.make_nice_id(self.pull_request_id)
     def __json__(self):
         return dict(
             revisions=self.revisions
+        )
     def url(self, **kwargs):
         canonical = kwargs.pop('canonical', None)
         import kallithea.lib.helpers as h
         b = self.org_ref_parts[1]
         if b != self.other_ref_parts[1]:
             s = '/_/' + b
         else:
             s = '/_/' + self.title
         kwargs['extra'] = urlreadable(s)
         if canonical:
             return h.canonical_url('pullrequest_show', repo_name=self.other_repo.repo_name,
                                    pull_request_id=self.pull_request_id, **kwargs)
         return h.url('pullrequest_show', repo_name=self.other_repo.repo_name,
                      pull_request_id=self.pull_request_id, **kwargs)
 class PullRequestReviewer(Base, BaseDbModel):
     __tablename__ = 'pull_request_reviewers'
     __table_args__ = (
         Index('pull_request_reviewers_user_id_idx', 'user_id'),
         _table_args_default_dict,
+    )
     def __init__(self, user=None, pull_request=None):
         self.user = user
         self.pull_request = pull_request
     pull_request_reviewers_id = Column('pull_requests_reviewers_id', Integer(), primary_key=True)
     pull_request_id = Column(Integer(), ForeignKey('pull_requests.pull_request_id'), nullable=False)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     user = relationship('User')
     pull_request = relationship('PullRequest')
 class Notification(Base, BaseDbModel):
     __tablename__ = 'notifications'
     __table_args__ = (
         Index('notification_type_idx', 'type'),
         _table_args_default_dict,
+    )
     TYPE_CHANGESET_COMMENT = u'cs_comment'
     TYPE_MESSAGE = u'message'
     TYPE_MENTION = u'mention' # not used
     TYPE_REGISTRATION = u'registration'
     TYPE_PULL_REQUEST = u'pull_request'
     TYPE_PULL_REQUEST_COMMENT = u'pull_request_comment'
     notification_id = Column(Integer(), primary_key=True)
     subject = Column(Unicode(512), nullable=False)
     body = Column(UnicodeText(), nullable=False)
     created_by = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     type_ = Column('type', Unicode(255), nullable=False)
     created_by_user = relationship('User')
     notifications_to_users = relationship('UserNotification', cascade="all, delete-orphan")
     @property
     def recipients(self):
         return [x.user for x in UserNotification.query()
                 .filter(UserNotification.notification == self)
                 .order_by(UserNotification.user_id.asc()).all()]
     @classmethod
     def create(cls, created_by, subject, body, recipients, type_=None):
         if type_ is None:
             type_ = Notification.TYPE_MESSAGE
         notification = cls()
         notification.created_by_user = created_by
         notification.subject = subject
         notification.body = body
         notification.type_ = type_
         notification.created_on = datetime.datetime.now()
         for recipient in recipients:
             un = UserNotification()
             un.notification = notification
             un.user_id = recipient.user_id
             # Mark notifications to self "pre-read" - should perhaps just be skipped
             if recipient == created_by:
                 un.read = True
             Session().add(un)
         Session().add(notification)
         Session().flush() # assign notification.notification_id
         return notification
     @property
     def description(self):
         from kallithea.model.notification import NotificationModel
         return NotificationModel().make_description(self)
 class UserNotification(Base, BaseDbModel):
     __tablename__ = 'user_to_notification'
     __table_args__ = (
         UniqueConstraint('user_id', 'notification_id'),
         _table_args_default_dict,
+    )
     user_id = Column(Integer(), ForeignKey('users.user_id'), primary_key=True)
     notification_id = Column(Integer(), ForeignKey('notifications.notification_id'), primary_key=True)
     read = Column(Boolean, nullable=False, default=False)
     sent_on = Column(DateTime(timezone=False), nullable=True) # FIXME: not nullable?
     user = relationship('User')
     notification = relationship('Notification')
     def mark_as_read(self):
         self.read = True
 class Gist(Base, BaseDbModel):
     __tablename__ = 'gists'
     __table_args__ = (
         Index('g_gist_access_id_idx', 'gist_access_id'),
         Index('g_created_on_idx', 'created_on'),
         _table_args_default_dict,
+    )
     GIST_PUBLIC = u'public'
     GIST_PRIVATE = u'private'
     DEFAULT_FILENAME = u'gistfile1.txt'
     gist_id = Column(Integer(), primary_key=True)
     gist_access_id = Column(Unicode(250), nullable=False)
     gist_description = Column(UnicodeText(), nullable=False)
     owner_id = Column('user_id', Integer(), ForeignKey('users.user_id'), nullable=False)
     gist_expires = Column(Float(53), nullable=False)
     gist_type = Column(Unicode(128), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     modified_at = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     owner = relationship('User')
     def __repr__(self):
         return '<Gist:[%s]%s>' % (self.gist_type, self.gist_access_id)
     @classmethod
     def guess_instance(cls, value):
         return super(Gist, cls).guess_instance(value, Gist.get_by_access_id)
     @classmethod
     def get_or_404(cls, id_):
         res = cls.query().filter(cls.gist_access_id == id_).scalar()
         if res is None:
             raise HTTPNotFound
         return res
     @classmethod
     def get_by_access_id(cls, gist_access_id):
         return cls.query().filter(cls.gist_access_id == gist_access_id).scalar()
     def gist_url(self):
         import kallithea
         alias_url = kallithea.CONFIG.get('gist_alias_url')
         if alias_url:
             return alias_url.replace('{gistid}', self.gist_access_id)
         import kallithea.lib.helpers as h
         return h.canonical_url('gist', gist_id=self.gist_access_id)
     @classmethod
     def base_path(cls):
         """
         Returns base path where all gists are stored
         :param cls:
         """
         from kallithea.model.gist import GIST_STORE_LOC
         q = Session().query(Ui) \
             .filter(Ui.ui_key == URL_SEP)
         q = q.options(FromCache("sql_cache_short", "repository_repo_path"))
         return os.path.join(q.one().ui_value, GIST_STORE_LOC)
     def get_api_data(self):
         """
         Common function for generating gist related data for API
         """
         gist = self
         data = dict(
             gist_id=gist.gist_id,
             type=gist.gist_type,
             access_id=gist.gist_access_id,
             description=gist.gist_description,
             url=gist.gist_url(),
             expires=gist.gist_expires,
             created_on=gist.created_on,
+        )
         return data
     def __json__(self):
         data = dict(
+        )
         data.update(self.get_api_data())
         return data
     ## SCM functions
     @property
     def scm_instance(self):
         from kallithea.lib.vcs import get_repo
         base_path = self.base_path()
         return get_repo(os.path.join(*map(safe_str,
                                           [base_path, self.gist_access_id])))

kallithea/model/gist.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.gist
 ~~~~~~~~~~~~~~~~~~~~
 gist model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: May 9, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import time
 import logging
 import traceback
 import shutil
 from kallithea.lib.utils2 import safe_unicode, unique_id, safe_int, \
     time_to_datetime, AttributeDict
 from kallithea.lib.compat import json
 from kallithea.model.base import BaseModel
 from kallithea.model.db import Gist
 from kallithea.model.repo import RepoModel
 from kallithea.model.scm import ScmModel
 log = logging.getLogger(__name__)
 GIST_STORE_LOC = '.rc_gist_store'
 GIST_METADATA_FILE = '.rc_gist_metadata'
 class GistModel(BaseModel):
     def _get_gist(self, gist):
         """
         Helper method to get gist by ID, or gist_access_id as a fallback
         :param gist: GistID, gist_access_id, or Gist instance
         """
-        return Gist.guess_instance(gist, callback=Gist.get_by_access_id)
         return Gist.guess_instance(gist)
     def __delete_gist(self, gist):
         """
         removes gist from filesystem
         :param gist: gist object
         """
         root_path = RepoModel().repos_path
         rm_path = os.path.join(root_path, GIST_STORE_LOC, gist.gist_access_id)
         log.info("Removing %s", rm_path)
         shutil.rmtree(rm_path)
     def _store_metadata(self, repo, gist_id, gist_access_id, user_id, gist_type,
                         gist_expires):
         """
         store metadata inside the gist, this can be later used for imports
         or gist identification
         """
         metadata = {
             'metadata_version': '1',
             'gist_db_id': gist_id,
             'gist_access_id': gist_access_id,
             'gist_owner_id': user_id,
             'gist_type': gist_type,
             'gist_expires': gist_expires,
             'gist_updated': time.time(),
+        }
         with open(os.path.join(repo.path, '.hg', GIST_METADATA_FILE), 'wb') as f:
             f.write(json.dumps(metadata))
     def get_gist(self, gist):
         return self._get_gist(gist)
     def get_gist_files(self, gist_access_id, revision=None):
         """
         Get files for given gist
         :param gist_access_id:
         """
         repo = Gist.get_by_access_id(gist_access_id)
         cs = repo.scm_instance.get_changeset(revision)
         return cs, [n for n in cs.get_node('/')]
     def create(self, description, owner, gist_mapping,
                gist_type=Gist.GIST_PUBLIC, lifetime=-1):
         """
         :param description: description of the gist
         :param owner: user who created this gist
         :param gist_mapping: mapping {filename:{'content':content},...}
         :param gist_type: type of gist private/public
         :param lifetime: in minutes, -1 == forever
         """
         owner = self._get_user(owner)
         gist_id = safe_unicode(unique_id(20))
         lifetime = safe_int(lifetime, -1)
         gist_expires = time.time() + (lifetime * 60) if lifetime != -1 else -1
         log.debug('set GIST expiration date to: %s',
                   time_to_datetime(gist_expires)
                    if gist_expires != -1 else 'forever')
         #create the Database version
         gist = Gist()
         gist.gist_description = description
         gist.gist_access_id = gist_id
         gist.owner_id = owner.user_id
         gist.gist_expires = gist_expires
         gist.gist_type = safe_unicode(gist_type)
         self.sa.add(gist)
         self.sa.flush() # make database assign gist.gist_id
         if gist_type == Gist.GIST_PUBLIC:
             # use DB ID for easy to use GIST ID
             gist_id = safe_unicode(gist.gist_id)
             gist.gist_access_id = gist_id
             self.sa.add(gist)
         gist_repo_path = os.path.join(GIST_STORE_LOC, gist_id)
         log.debug('Creating new %s GIST repo in %s', gist_type, gist_repo_path)
         repo = RepoModel()._create_filesystem_repo(
             repo_name=gist_id, repo_type='hg', repo_group=GIST_STORE_LOC)
         processed_mapping = {}
         for filename in gist_mapping:
             if filename != os.path.basename(filename):
                 raise Exception('Filename cannot be inside a directory')
             content = gist_mapping[filename]['content']
             #TODO: expand support for setting explicit lexers
 #             if lexer is None:
 #                 try:
 #                     guess_lexer = pygments.lexers.guess_lexer_for_filename
 #                     lexer = guess_lexer(filename,content)
 #                 except pygments.util.ClassNotFound:
 #                     lexer = 'text'
             processed_mapping[filename] = {'content': content}
         # now create single multifile commit
         message = 'added file'
         message += 's: ' if len(processed_mapping) > 1 else ': '
         message += ', '.join([x for x in processed_mapping])
         #fake Kallithea Repository object
         fake_repo = AttributeDict(dict(
             repo_name=gist_repo_path,
             scm_instance_no_cache=lambda: repo,
         ))
         ScmModel().create_nodes(
             user=owner.user_id, repo=fake_repo,
             message=message,
             nodes=processed_mapping,
             trigger_push_hook=False
+        )
         self._store_metadata(repo, gist.gist_id, gist.gist_access_id,
                              owner.user_id, gist.gist_type, gist.gist_expires)
         return gist
     def delete(self, gist, fs_remove=True):
         gist = self._get_gist(gist)
         try:
             self.sa.delete(gist)
             if fs_remove:
                 self.__delete_gist(gist)
             else:
                 log.debug('skipping removal from filesystem')
         except Exception:
             log.error(traceback.format_exc())
             raise
     def update(self, gist, description, owner, gist_mapping, gist_type,
                lifetime):
         gist = self._get_gist(gist)
         gist_repo = gist.scm_instance
         lifetime = safe_int(lifetime, -1)
         if lifetime == 0:  # preserve old value
             gist_expires = gist.gist_expires
         else:
             gist_expires = time.time() + (lifetime * 60) if lifetime != -1 else -1
         #calculate operation type based on given data
         gist_mapping_op = {}
         for k, v in gist_mapping.items():
             # add, mod, del
             if not v['org_filename'] and v['filename']:
                 op = 'add'
             elif v['org_filename'] and not v['filename']:
                 op = 'del'
             else:
                 op = 'mod'
             v['op'] = op
             gist_mapping_op[k] = v
         gist.gist_description = description
         gist.gist_expires = gist_expires
         gist.owner = owner
         gist.gist_type = gist_type
         message = 'updated file'
         message += 's: ' if len(gist_mapping) > 1 else ': '
         message += ', '.join([x for x in gist_mapping])
         #fake Kallithea Repository object
         fake_repo = AttributeDict(dict(
             repo_name=gist_repo.path,
             scm_instance_no_cache=lambda: gist_repo,
         ))
         self._store_metadata(gist_repo, gist.gist_id, gist.gist_access_id,
                              owner.user_id, gist.gist_type, gist.gist_expires)
         ScmModel().update_nodes(
             user=owner.user_id,
             repo=fake_repo,
             message=message,
             nodes=gist_mapping_op,
             trigger_push_hook=False
+        )
         return gist

kallithea/model/repo.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.repo
 ~~~~~~~~~~~~~~~~~~~~
 Repository model for kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jun 5, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import shutil
 import logging
 import traceback
 from datetime import datetime
 from sqlalchemy.orm import subqueryload
 from kallithea.lib.utils import make_ui
 from kallithea.lib.vcs.backends import get_backend
 from kallithea.lib.compat import json
 from kallithea.lib.utils2 import LazyProperty, safe_str, safe_unicode, \
     remove_prefix, obfuscate_url_pw, get_current_authuser
 from kallithea.lib.caching_query import FromCache
 from kallithea.lib.hooks import log_delete_repository
 from kallithea.model.base import BaseModel
 from kallithea.model.db import Repository, UserRepoToPerm, UserGroupRepoToPerm, \
     UserRepoGroupToPerm, UserGroupRepoGroupToPerm, User, Permission, \
     Statistics, UserGroup, Ui, RepoGroup, RepositoryField
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import HasRepoPermissionAny, HasUserGroupPermissionAny
 from kallithea.lib.exceptions import AttachedForksError
 from kallithea.model.scm import UserGroupList
 log = logging.getLogger(__name__)
 class RepoModel(BaseModel):
     URL_SEPARATOR = Repository.url_sep()
     def _get_user_group(self, users_group):
         return UserGroup.guess_instance(users_group,
                                   callback=UserGroup.get_by_group_name)
         return UserGroup.guess_instance(users_group)
     def _get_repo_group(self, repo_group):
         return RepoGroup.guess_instance(repo_group,
                                   callback=RepoGroup.get_by_group_name)
         return RepoGroup.guess_instance(repo_group)
     def _create_default_perms(self, repository, private):
         # create default permission
         default = 'repository.read'
         def_user = User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('repository.'):
                 default = p.permission.permission_name
                 break
         default_perm = 'repository.none' if private else default
         repo_to_perm = UserRepoToPerm()
         repo_to_perm.permission = Permission.get_by_key(default_perm)
         repo_to_perm.repository = repository
         repo_to_perm.user_id = def_user.user_id
         return repo_to_perm
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = self.sa.query(Ui).filter(Ui.ui_key == '/').one()
         return q.ui_value
     def get(self, repo_id, cache=False):
         repo = self.sa.query(Repository) \
             .filter(Repository.repo_id == repo_id)
         if cache:
             repo = repo.options(FromCache("sql_cache_short",
                                           "get_repo_%s" % repo_id))
         return repo.scalar()
     def get_repo(self, repository):
         return self._get_repo(repository)
     def get_by_repo_name(self, repo_name, cache=False):
         repo = self.sa.query(Repository) \
             .filter(Repository.repo_name == repo_name)
         if cache:
             repo = repo.options(FromCache("sql_cache_short",
                                           "get_repo_%s" % repo_name))
         return repo.scalar()
     def get_all_user_repos(self, user):
         """
         Gets all repositories that user have at least read access
         :param user:
         """
         from kallithea.lib.auth import AuthUser
         user = self._get_user(user)
         repos = AuthUser(dbuser=user).permissions['repositories']
         access_check = lambda r: r[1] in ['repository.read',
                                           'repository.write',
                                           'repository.admin']
         repos = [x[0] for x in filter(access_check, repos.items())]
         return Repository.query().filter(Repository.repo_name.in_(repos))
     def get_users_js(self):
         users = self.sa.query(User) \
             .filter(User.active == True) \
             .order_by(User.name, User.lastname) \
             .all()
         return json.dumps([
+            {
                 'id': u.user_id,
                 'fname': h.escape(u.name),
                 'lname': h.escape(u.lastname),
                 'nname': u.username,
                 'gravatar_lnk': h.gravatar_url(u.email, size=28, default='default'),
                 'gravatar_size': 14,
             } for u in users]
+        )
     def get_user_groups_js(self):
         user_groups = self.sa.query(UserGroup) \
             .filter(UserGroup.users_group_active == True) \
             .order_by(UserGroup.users_group_name) \
             .options(subqueryload(UserGroup.members)) \
             .all()
         user_groups = UserGroupList(user_groups, perm_set=['usergroup.read',
                                                            'usergroup.write',
                                                            'usergroup.admin'])
         return json.dumps([
+            {
                 'id': gr.users_group_id,
                 'grname': gr.users_group_name,
                 'grmembers': len(gr.members),
             } for gr in user_groups]
+        )
     @classmethod
     def _render_datatable(cls, tmpl, *args, **kwargs):
         import kallithea
         from pylons import tmpl_context as c, request
         from pylons.i18n.translation import _
         _tmpl_lookup = kallithea.CONFIG['pylons.app_globals'].mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         tmpl = template.get_def(tmpl)
         kwargs.update(dict(_=_, h=h, c=c, request=request))
         return tmpl.render(*args, **kwargs)
     def get_repos_as_dict(self, repos_list=None, admin=False, perm_check=True,
                           super_user_actions=False, short_name=False):
         _render = self._render_datatable
         from pylons import tmpl_context as c
         def repo_lnk(name, rtype, rstate, private, fork_of):
             return _render('repo_name', name, rtype, rstate, private, fork_of,
                            short_name=short_name, admin=False)
         def last_change(last_change):
             return _render("last_change", last_change)
         def rss_lnk(repo_name):
             return _render("rss", repo_name)
         def atom_lnk(repo_name):
             return _render("atom", repo_name)
         def last_rev(repo_name, cs_cache):
             return _render('revision', repo_name, cs_cache.get('revision'),
                            cs_cache.get('raw_id'), cs_cache.get('author'),
                            cs_cache.get('message'))
         def desc(desc):
             return h.urlify_text(desc, truncate=80, stylize=c.visual.stylify_metatags)
         def state(repo_state):
             return _render("repo_state", repo_state)
         def repo_actions(repo_name):
             return _render('repo_actions', repo_name, super_user_actions)
         def owner_actions(owner_id, username):
             return _render('user_name', owner_id, username)
         repos_data = []
         for repo in repos_list:
             if perm_check:
                 # check permission at this level
                 if not HasRepoPermissionAny(
                         'repository.read', 'repository.write',
                         'repository.admin'
                 )(repo.repo_name, 'get_repos_as_dict check'):
                     continue
             cs_cache = repo.changeset_cache
             row = {
                 "raw_name": repo.repo_name,
                 "just_name": repo.just_name,
                 "name": repo_lnk(repo.repo_name, repo.repo_type,
                                  repo.repo_state, repo.private, repo.fork),
                 "last_change_iso": repo.last_db_change.isoformat(),
                 "last_change": last_change(repo.last_db_change),
                 "last_changeset": last_rev(repo.repo_name, cs_cache),
                 "last_rev_raw": cs_cache.get('revision'),
                 "desc": desc(repo.description),
                 "owner": h.person(repo.owner),
                 "state": state(repo.repo_state),
                 "rss": rss_lnk(repo.repo_name),
                 "atom": atom_lnk(repo.repo_name),
+            }
             if admin:
                 row.update({
                     "action": repo_actions(repo.repo_name),
                     "owner": owner_actions(repo.owner_id,
                                            h.person(repo.owner))
                 })
             repos_data.append(row)
         return {
             "totalRecords": len(repos_list),
             "startIndex": 0,
             "sort": "name",
             "dir": "asc",
             "records": repos_data
+        }
     def _get_defaults(self, repo_name):
         """
         Gets information about repository, and returns a dict for
         usage in forms

kallithea/model/repo_group.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.repo_group
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 repo group model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jan 25, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import logging
 import traceback
 import shutil
 import datetime
 from kallithea.lib.utils2 import LazyProperty
 from kallithea.model.base import BaseModel
 from kallithea.model.db import RepoGroup, Ui, UserRepoGroupToPerm, \
     User, Permission, UserGroupRepoGroupToPerm, UserGroup, Repository
 log = logging.getLogger(__name__)
 class RepoGroupModel(BaseModel):
     def _get_user_group(self, users_group):
         return UserGroup.guess_instance(users_group,
                                   callback=UserGroup.get_by_group_name)
         return UserGroup.guess_instance(users_group)
     def _get_repo_group(self, repo_group):
         return RepoGroup.guess_instance(repo_group,
                                   callback=RepoGroup.get_by_group_name)
         return RepoGroup.guess_instance(repo_group)
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = Ui.get_by_key('paths', '/')
         return q.ui_value
     def _create_default_perms(self, new_group):
         # create default permission
         default_perm = 'group.read'
         def_user = User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('group.'):
                 default_perm = p.permission.permission_name
                 break
         repo_group_to_perm = UserRepoGroupToPerm()
         repo_group_to_perm.permission = Permission.get_by_key(default_perm)
         repo_group_to_perm.group = new_group
         repo_group_to_perm.user_id = def_user.user_id
         return repo_group_to_perm
     def _create_group(self, group_name):
         """
         makes repository group on filesystem
         :param repo_name:
         :param parent_id:
         """
         create_path = os.path.join(self.repos_path, group_name)
         log.debug('creating new group in %s', create_path)
         if os.path.isdir(create_path):
             raise Exception('That directory already exists !')
         os.makedirs(create_path)
         log.debug('Created group in %s', create_path)
     def _rename_group(self, old, new):
         """
         Renames a group on filesystem
         :param group_name:
         """
         if old == new:
             log.debug('skipping group rename')
             return
         log.debug('renaming repository group from %s to %s', old, new)
         old_path = os.path.join(self.repos_path, old)
         new_path = os.path.join(self.repos_path, new)
         log.debug('renaming repos paths from %s to %s', old_path, new_path)
         if os.path.isdir(new_path):
             raise Exception('Was trying to rename to already '
                             'existing dir %s' % new_path)
         shutil.move(old_path, new_path)
     def _delete_group(self, group, force_delete=False):
         """
         Deletes a group from a filesystem
         :param group: instance of group from database
         :param force_delete: use shutil rmtree to remove all objects
         """
         paths = group.full_path.split(RepoGroup.url_sep())
         paths = os.sep.join(paths)
         rm_path = os.path.join(self.repos_path, paths)
         log.info("Removing group %s", rm_path)
         # delete only if that path really exists
         if os.path.isdir(rm_path):
             if force_delete:
                 shutil.rmtree(rm_path)
             else:
                 #archive that group`
                 _now = datetime.datetime.now()
                 _ms = str(_now.microsecond).rjust(6, '0')
                 _d = 'rm__%s_GROUP_%s' % (_now.strftime('%Y%m%d_%H%M%S_' + _ms),
                                           group.name)
                 shutil.move(rm_path, os.path.join(self.repos_path, _d))
     def create(self, group_name, group_description, owner, parent=None,
                just_db=False, copy_permissions=False):
         try:
             owner = self._get_user(owner)
             parent_group = self._get_repo_group(parent)
             new_repo_group = RepoGroup()
             new_repo_group.owner = owner
             new_repo_group.group_description = group_description or group_name
             new_repo_group.parent_group = parent_group
             new_repo_group.group_name = new_repo_group.get_new_name(group_name)
             self.sa.add(new_repo_group)
             # create an ADMIN permission for owner except if we're super admin,
             # later owner should go into the owner field of groups
             if not owner.is_admin:
                 self.grant_user_permission(repo_group=new_repo_group,
                                            user=owner, perm='group.admin')
             if parent_group and copy_permissions:
                 # copy permissions from parent
                 user_perms = UserRepoGroupToPerm.query() \
                     .filter(UserRepoGroupToPerm.group == parent_group).all()
                 group_perms = UserGroupRepoGroupToPerm.query() \
                     .filter(UserGroupRepoGroupToPerm.group == parent_group).all()
                 for perm in user_perms:
                     # don't copy over the permission for user who is creating
                     # this group, if he is not super admin he get's admin
                     # permission set above
                     if perm.user != owner or owner.is_admin:
                         UserRepoGroupToPerm.create(perm.user, new_repo_group, perm.permission)
                 for perm in group_perms:
                     UserGroupRepoGroupToPerm.create(perm.users_group, new_repo_group, perm.permission)
             else:
                 perm_obj = self._create_default_perms(new_repo_group)
                 self.sa.add(perm_obj)
             if not just_db:
                 # we need to flush here, in order to check if database won't
                 # throw any exceptions, create filesystem dirs at the very end
                 self.sa.flush()
                 self._create_group(new_repo_group.group_name)
             return new_repo_group
         except Exception:
             log.error(traceback.format_exc())
             raise
     def _update_permissions(self, repo_group, perms_new=None,
                             perms_updates=None, recursive=None,
                             check_perms=True):
         from kallithea.model.repo import RepoModel
         from kallithea.lib.auth import HasUserGroupPermissionAny
         if not perms_new:
             perms_new = []
         if not perms_updates:
             perms_updates = []
         def _set_perm_user(obj, user, perm):
             if isinstance(obj, RepoGroup):
                 self.grant_user_permission(repo_group=obj, user=user, perm=perm)
             elif isinstance(obj, Repository):
                 # private repos will not allow to change the default permissions
                 # using recursive mode
                 if obj.private and user == User.DEFAULT_USER:
                     return
                 # we set group permission but we have to switch to repo
                 # permission
                 perm = perm.replace('group.', 'repository.')
                 RepoModel().grant_user_permission(
                     repo=obj, user=user, perm=perm
+                )
         def _set_perm_group(obj, users_group, perm):
             if isinstance(obj, RepoGroup):
                 self.grant_user_group_permission(repo_group=obj,
                                                   group_name=users_group,
                                                   perm=perm)
             elif isinstance(obj, Repository):
                 # we set group permission but we have to switch to repo
                 # permission
                 perm = perm.replace('group.', 'repository.')
                 RepoModel().grant_user_group_permission(
                     repo=obj, group_name=users_group, perm=perm
+                )
         # start updates
         updates = []
         log.debug('Now updating permissions for %s in recursive mode:%s',
                   repo_group, recursive)
         for obj in repo_group.recursive_groups_and_repos():
             # iterated obj is an instance of a repos group or repository in
             # that group, recursive option can be: none, repos, groups, all
             if recursive == 'all':
                 pass
             elif recursive == 'repos':

kallithea/model/user_group.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.user_group
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 user group model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Oct 1, 2011
 :author: nvinot, marcink
 """
 import logging
 import traceback
 from kallithea.model.base import BaseModel
 from kallithea.model.db import UserGroupMember, UserGroup, \
     UserGroupRepoToPerm, Permission, UserGroupToPerm, User, UserUserGroupToPerm, \
     UserGroupUserGroupToPerm
 from kallithea.lib.exceptions import UserGroupsAssignedException, \
     RepoGroupAssignmentError
 log = logging.getLogger(__name__)
 class UserGroupModel(BaseModel):
     def _get_user_group(self, user_group):
         return UserGroup.guess_instance(user_group,
                                   callback=UserGroup.get_by_group_name)
         return UserGroup.guess_instance(user_group)
     def _create_default_perms(self, user_group):
         # create default permission
         default_perm = 'usergroup.read'
         def_user = User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('usergroup.'):
                 default_perm = p.permission.permission_name
                 break
         user_group_to_perm = UserUserGroupToPerm()
         user_group_to_perm.permission = Permission.get_by_key(default_perm)
         user_group_to_perm.user_group = user_group
         user_group_to_perm.user_id = def_user.user_id
         return user_group_to_perm
     def _update_permissions(self, user_group, perms_new=None,
                             perms_updates=None):
         from kallithea.lib.auth import HasUserGroupPermissionAny
         if not perms_new:
             perms_new = []
         if not perms_updates:
             perms_updates = []
         # update permissions
         for member, perm, member_type in perms_updates:
             if member_type == 'user':
                 # this updates existing one
                 self.grant_user_permission(
                     user_group=user_group, user=member, perm=perm
+                )
             else:
                 #check if we have permissions to alter this usergroup
                 if HasUserGroupPermissionAny('usergroup.read', 'usergroup.write',
                                              'usergroup.admin')(member):
                     self.grant_user_group_permission(
                         target_user_group=user_group, user_group=member, perm=perm
+                    )
         # set new permissions
         for member, perm, member_type in perms_new:
             if member_type == 'user':
                 self.grant_user_permission(
                     user_group=user_group, user=member, perm=perm
+                )
             else:
                 #check if we have permissions to alter this usergroup
                 if HasUserGroupPermissionAny('usergroup.read', 'usergroup.write',
                                              'usergroup.admin')(member):
                     self.grant_user_group_permission(
                         target_user_group=user_group, user_group=member, perm=perm
+                    )
     def get(self, user_group_id, cache=False):
         return UserGroup.get(user_group_id)
     def get_group(self, user_group):
         return self._get_user_group(user_group)
     def get_by_name(self, name, cache=False, case_insensitive=False):
         return UserGroup.get_by_group_name(name, cache, case_insensitive)
     def create(self, name, description, owner, active=True, group_data=None):
         try:
             new_user_group = UserGroup()
             new_user_group.owner = self._get_user(owner)
             new_user_group.users_group_name = name
             new_user_group.user_group_description = description
             new_user_group.users_group_active = active
             if group_data:
                 new_user_group.group_data = group_data
             self.sa.add(new_user_group)
             perm_obj = self._create_default_perms(new_user_group)
             self.sa.add(perm_obj)
             self.grant_user_permission(user_group=new_user_group,
                                        user=owner, perm='usergroup.admin')
             return new_user_group
         except Exception:
             log.error(traceback.format_exc())
             raise
     def update(self, user_group, form_data):
         try:
             user_group = self._get_user_group(user_group)
             for k, v in form_data.items():
                 if k == 'users_group_members':
                     members_list = []
                     if v:
                         v = [v] if isinstance(v, basestring) else v
                         for u_id in set(v):
                             member = UserGroupMember(user_group.users_group_id, u_id)
                             members_list.append(member)
                             self.sa.add(member)
                     user_group.members = members_list
                 setattr(user_group, k, v)
             # Flush to make db assign users_group_member_id to newly
             # created UserGroupMembers.
             self.sa.flush()
         except Exception:
             log.error(traceback.format_exc())
             raise
     def delete(self, user_group, force=False):
         """
         Deletes user group, unless force flag is used
         raises exception if there are members in that group, else deletes
         group and users
         :param user_group:
         :param force:
         """
         user_group = self._get_user_group(user_group)
         try:
             # check if this group is not assigned to repo
             assigned_groups = UserGroupRepoToPerm.query() \
                 .filter(UserGroupRepoToPerm.users_group == user_group).all()
             assigned_groups = [x.repository.repo_name for x in assigned_groups]
             if assigned_groups and not force:
                 raise UserGroupsAssignedException(
                     'User Group assigned to %s' % ", ".join(assigned_groups))
             self.sa.delete(user_group)
         except Exception:
             log.error(traceback.format_exc())
             raise
     def add_user_to_group(self, user_group, user):
         user_group = self._get_user_group(user_group)
         user = self._get_user(user)
         for m in user_group.members:
             u = m.user
             if u.user_id == user.user_id:
                 # user already in the group, skip
                 return True
         try:
             user_group_member = UserGroupMember()
             user_group_member.user = user
             user_group_member.users_group = user_group
             user_group.members.append(user_group_member)
             user.group_member.append(user_group_member)
             self.sa.add(user_group_member)
             return user_group_member
         except Exception:
             log.error(traceback.format_exc())
             raise
     def remove_user_from_group(self, user_group, user):
         user_group = self._get_user_group(user_group)
         user = self._get_user(user)
         user_group_member = None
         for m in user_group.members:
             if m.user_id == user.user_id:
                 # Found this user's membership row
                 user_group_member = m
                 break
         if user_group_member:
             try:
                 self.sa.delete(user_group_member)
                 return True
             except Exception:
                 log.error(traceback.format_exc())
                 raise
         else:
             # User isn't in that group
             return False
     def has_perm(self, user_group, perm):
         user_group = self._get_user_group(user_group)
         perm = self._get_perm(perm)
         return UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == user_group) \
             .filter(UserGroupToPerm.permission == perm).scalar() is not None
     def grant_perm(self, user_group, perm):
         user_group = self._get_user_group(user_group)
         perm = self._get_perm(perm)
         # if this permission is already granted skip it
         _perm = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == user_group) \

0 comments (0 inline, 0 general)