# -*- coding: utf-8 -*-
"""
rhodecode.controllers.login
~~~~~~~~~~~~~~~~~~~~~~~~~~~
Login controller for rhodeocode
:created_on: Apr 22, 2010
:author: marcink
:copyright: (C) 2010-2012 Marcin Kuzminski <marcin@python-works.com>
:license: GPLv3, see COPYING for more details.
# 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/>.
import logging
import formencode
import datetime
import urlparse
from formencode import htmlfill
from webob.exc import HTTPFound
from pylons.i18n.translation import _
from pylons.controllers.util import abort, redirect
from pylons import request, response, session, tmpl_context as c, url
import rhodecode.lib.helpers as h
from rhodecode.lib.auth import AuthUser, HasPermissionAnyDecorator
from rhodecode.lib.base import BaseController, render
from rhodecode.model.db import User
from rhodecode.model.forms import LoginForm, RegisterForm, PasswordResetForm
from rhodecode.model.user import UserModel
from rhodecode.model.meta import Session
log = logging.getLogger(__name__)
class LoginController(BaseController):
def __before__(self):
super(LoginController, self).__before__()
def index(self):
# redirect if already logged in
c.came_from = request.GET.get('came_from')
not_default = self.rhodecode_user.username != 'default'
ip_allowed = self.rhodecode_user.ip_allowed
if self.rhodecode_user.is_authenticated and not_default and ip_allowed:
return redirect(url('home'))
if request.POST:
# import Login Form validator class
login_form = LoginForm()
try:
session.invalidate()
c.form_result = login_form.to_python(dict(request.POST))
# form checks for username/password, now we're authenticated
username = c.form_result['username']
user = User.get_by_username(username, case_insensitive=True)
auth_user = AuthUser(user.user_id)
auth_user.set_authenticated()
cs = auth_user.get_cookie_store()
session['rhodecode_user'] = cs
user.update_lastlogin()
Session().commit()
# If they want to be remembered, update the cookie
if c.form_result['remember'] is not False:
_year = (datetime.datetime.now() +
datetime.timedelta(seconds=60 * 60 * 24 * 365))
session._set_cookie_expires(_year)
session.save()
log.info('user %s is now authenticated and stored in '
'session, session attrs %s' % (username, cs))
# dumps session attrs back to cookie
session._update_cookie_out()
# we set new cookie
headers = None
if session.request['set_cookie']:
# send set-cookie headers back to response to update cookie
headers = [('Set-Cookie', session.request['cookie_out'])]
allowed_schemes = ['http', 'https']
if c.came_from:
parsed = urlparse.urlparse(c.came_from)
server_parsed = urlparse.urlparse(url.current())
if parsed.scheme and parsed.scheme not in allowed_schemes:
log.error(
'Suspicious URL scheme detected %s for url %s' %
(parsed.scheme, parsed))
c.came_from = url('home')
elif server_parsed.netloc != parsed.netloc:
log.error('Suspicious NETLOC detected %s for url %s'
'server url is: %s' %
(parsed.netloc, parsed, server_parsed))
raise HTTPFound(location=c.came_from, headers=headers)
else:
raise HTTPFound(location=url('home'), headers=headers)
except formencode.Invalid, errors:
return htmlfill.render(
render('/login.html'),
defaults=errors.value,
errors=errors.error_dict or {},
prefix_error=False,
encoding="UTF-8")
return render('/login.html')
@HasPermissionAnyDecorator('hg.admin', 'hg.register.auto_activate',
'hg.register.manual_activate')
def register(self):
c.auto_active = False
for perm in User.get_by_username('default').user_perms:
if perm.permission.permission_name == 'hg.register.auto_activate':
c.auto_active = True
break
c.auto_active = 'hg.register.auto_activate' in User.get_by_username('default')\
.AuthUser.permissions['global']
register_form = RegisterForm()()
form_result = register_form.to_python(dict(request.POST))
form_result['active'] = c.auto_active
UserModel().create_registration(form_result)
h.flash(_('You have successfully registered into rhodecode'),
h.flash(_('You have successfully registered into RhodeCode'),
category='success')
return redirect(url('login_home'))
render('/register.html'),
return render('/register.html')
def password_reset(self):
password_reset_form = PasswordResetForm()()
form_result = password_reset_form.to_python(dict(request.POST))
UserModel().reset_password_link(form_result)
h.flash(_('Your password reset link was sent'),
render('/password_reset.html'),
return render('/password_reset.html')
def password_reset_confirmation(self):
if request.GET and request.GET.get('key'):
user = User.get_by_api_key(request.GET.get('key'))
data = dict(email=user.email)
UserModel().reset_password(data)
h.flash(_('Your password reset was successful, '
'new password has been sent to your email'),
except Exception, e:
log.error(e)
return redirect(url('reset_password'))
def logout(self):
session.delete()
log.info('Logging out and deleting session for user')
redirect(url('home'))
rhodecode.lib.auth
~~~~~~~~~~~~~~~~~~
authentication and permission libraries
:created_on: Apr 4, 2010
import random
import traceback
import hashlib
from tempfile import _RandomNameSequence
from decorator import decorator
from pylons import config, url, request
from sqlalchemy.orm.exc import ObjectDeletedError
from rhodecode import __platform__, is_windows, is_unix
from rhodecode.lib.utils2 import str2bool, safe_unicode
from rhodecode.lib.exceptions import LdapPasswordError, LdapUsernameError
from rhodecode.lib.utils import get_repo_slug, get_repos_group_slug
from rhodecode.lib.auth_ldap import AuthLdap
from rhodecode.model import meta
from rhodecode.model.db import Permission, RhodeCodeSetting, User, UserIpMap
from rhodecode.lib.caching_query import FromCache
class PasswordGenerator(object):
This is a simple class for generating password from different sets of
characters
usage::
passwd_gen = PasswordGenerator()
#print 8-letter password containing only big and small letters
of alphabet
passwd_gen.gen_password(8, passwd_gen.ALPHABETS_BIG_SMALL)
ALPHABETS_NUM = r'''1234567890'''
ALPHABETS_SMALL = r'''qwertyuiopasdfghjklzxcvbnm'''
ALPHABETS_BIG = r'''QWERTYUIOPASDFGHJKLZXCVBNM'''
ALPHABETS_SPECIAL = r'''`-=[]\;',./~!@#$%^&*()_+{}|:"<>?'''
ALPHABETS_FULL = ALPHABETS_BIG + ALPHABETS_SMALL \
+ ALPHABETS_NUM + ALPHABETS_SPECIAL
ALPHABETS_ALPHANUM = ALPHABETS_BIG + ALPHABETS_SMALL + ALPHABETS_NUM
ALPHABETS_BIG_SMALL = ALPHABETS_BIG + ALPHABETS_SMALL
ALPHABETS_ALPHANUM_BIG = ALPHABETS_BIG + ALPHABETS_NUM
ALPHABETS_ALPHANUM_SMALL = ALPHABETS_SMALL + ALPHABETS_NUM
def __init__(self, passwd=''):
self.passwd = passwd
def gen_password(self, length, type_=None):
if type_ is None:
type_ = self.ALPHABETS_FULL
self.passwd = ''.join([random.choice(type_) for _ in xrange(length)])
return self.passwd
class RhodeCodeCrypto(object):
@classmethod
def hash_string(cls, str_):
Cryptographic function used for password hashing based on pybcrypt
or pycrypto in windows
:param password: password to hash
if is_windows:
from hashlib import sha256
return sha256(str_).hexdigest()
elif is_unix:
import bcrypt
return bcrypt.hashpw(str_, bcrypt.gensalt(10))
raise Exception('Unknown or unsupported platform %s' \
% __platform__)
def hash_check(cls, password, hashed):
Checks matching password with it's hashed value, runs different
implementation based on platform it runs on
:param password: password
:param hashed: password in hashed form
return sha256(password).hexdigest() == hashed
return bcrypt.hashpw(password, hashed) == hashed
def get_crypt_password(password):
return RhodeCodeCrypto.hash_string(password)
def check_password(password, hashed):
return RhodeCodeCrypto.hash_check(password, hashed)
def generate_api_key(str_, salt=None):
Generates API KEY from given string
:param str_:
:param salt:
if salt is None:
salt = _RandomNameSequence().next()
return hashlib.sha1(str_ + salt).hexdigest()
def authfunc(environ, username, password):
Dummy authentication wrapper function used in Mercurial and Git for
access control.
:param environ: needed only for using in Basic auth
return authenticate(username, password)
def authenticate(username, password):
Authentication function used for access control,
firstly checks for db authentication then if ldap is enabled for ldap
authentication, also creates ldap user if not in database
:param username: username
user_model = UserModel()
user = User.get_by_username(username)
log.debug('Authenticating user using RhodeCode account')
if user is not None and not user.ldap_dn:
if user.active:
if user.username == 'default' and user.active:
log.info('user %s authenticated correctly as anonymous user' %
username)
return True
elif user.username == username and check_password(password,
user.password):
log.info('user %s authenticated correctly' % username)
log.warning('user %s tried auth but is disabled' % username)
log.debug('Regular authentication failed')
user_obj = User.get_by_username(username, case_insensitive=True)
if user_obj is not None and not user_obj.ldap_dn:
log.debug('this user already exists as non ldap')
return False
ldap_settings = RhodeCodeSetting.get_ldap_settings()
#======================================================================
# FALLBACK TO LDAP AUTH IF ENABLE
if str2bool(ldap_settings.get('ldap_active')):
log.debug("Authenticating user using ldap")
kwargs = {
'server': ldap_settings.get('ldap_host', ''),
'base_dn': ldap_settings.get('ldap_base_dn', ''),
'port': ldap_settings.get('ldap_port'),
'bind_dn': ldap_settings.get('ldap_dn_user'),
'bind_pass': ldap_settings.get('ldap_dn_pass'),
'tls_kind': ldap_settings.get('ldap_tls_kind'),
'tls_reqcert': ldap_settings.get('ldap_tls_reqcert'),
'ldap_filter': ldap_settings.get('ldap_filter'),
'search_scope': ldap_settings.get('ldap_search_scope'),
'attr_login': ldap_settings.get('ldap_attr_login'),
'ldap_version': 3,
}
log.debug('Checking for ldap authentication')
aldap = AuthLdap(**kwargs)
(user_dn, ldap_attrs) = aldap.authenticate_ldap(username,
password)
log.debug('Got ldap DN response %s' % user_dn)
get_ldap_attr = lambda k: ldap_attrs.get(ldap_settings\
.get(k), [''])[0]
user_attrs = {
'name': safe_unicode(get_ldap_attr('ldap_attr_firstname')),
'lastname': safe_unicode(get_ldap_attr('ldap_attr_lastname')),
'email': get_ldap_attr('ldap_attr_email'),
'active': 'hg.register.auto_activate' in User\
.get_by_username('default').AuthUser.permissions['global']
# don't store LDAP password since we don't need it. Override
# with some random generated password
_password = PasswordGenerator().gen_password(length=8)
# create this user on the fly if it doesn't exist in rhodecode
# database
if user_model.create_ldap(username, _password, user_dn,
user_attrs):
log.info('created new ldap user %s' % username)
except (LdapUsernameError, LdapPasswordError,):
pass
except (Exception,):
log.error(traceback.format_exc())
def login_container_auth(username):
if user is None:
'name': username,
'lastname': None,
'email': None,
user = UserModel().create_for_container_auth(username, user_attrs)
if not user:
return None
log.info('User %s was created by container authentication' % username)
if not user.active:
log.debug('User %s is now logged in by container authentication',
user.username)
return user
def get_container_username(environ, config, clean_username=False):
Get's the container_auth username (or email). It tries to get username
from REMOTE_USER if container_auth_enabled is enabled, if that fails
it tries to get username from HTTP_X_FORWARDED_USER if proxypass_auth_enabled
is enabled. clean_username extracts the username from this data if it's
having @ in it.
:param environ:
:param config:
:param clean_username:
username = None
if str2bool(config.get('container_auth_enabled', False)):
from paste.httpheaders import REMOTE_USER
username = REMOTE_USER(environ)
log.debug('extracted REMOTE_USER:%s' % (username))
if not username and str2bool(config.get('proxypass_auth_enabled', False)):
username = environ.get('HTTP_X_FORWARDED_USER')
log.debug('extracted HTTP_X_FORWARDED_USER:%s' % (username))
if username and clean_username:
# Removing realm and domain from username
username = username.partition('@')[0]
username = username.rpartition('\\')[2]
log.debug('Received username %s from container' % username)
return username
class CookieStoreWrapper(object):
def __init__(self, cookie_store):
self.cookie_store = cookie_store
def __repr__(self):
return 'CookieStore<%s>' % (self.cookie_store)
def get(self, key, other=None):
if isinstance(self.cookie_store, dict):
return self.cookie_store.get(key, other)
elif isinstance(self.cookie_store, AuthUser):
return self.cookie_store.__dict__.get(key, other)
class AuthUser(object):
A simple object that handles all attributes of user in RhodeCode
It does lookup based on API key,given user, or user present in session
Then it fills all required information for such user. It also checks if
anonymous access is enabled and if so, it returns default user as logged
in
def __init__(self, user_id=None, api_key=None, username=None, ip_addr=None):
self.user_id = user_id
self.api_key = None
self.username = username
self.ip_addr = ip_addr
self.name = ''
self.lastname = ''
self.email = ''
self.is_authenticated = False
self.admin = False
self.inherit_default_permissions = False
self.permissions = {}
self._api_key = api_key
self.propagate_data()
self._instance = None
def propagate_data(self):
self.anonymous_user = User.get_by_username('default', cache=True)
is_user_loaded = False
# try go get user by api key
if self._api_key and self._api_key != self.anonymous_user.api_key:
log.debug('Auth User lookup by API KEY %s' % self._api_key)
is_user_loaded = user_model.fill_data(self, api_key=self._api_key)
# lookup by userid
elif (self.user_id is not None and
self.user_id != self.anonymous_user.user_id):
log.debug('Auth User lookup by USER ID %s' % self.user_id)
is_user_loaded = user_model.fill_data(self, user_id=self.user_id)
# lookup by username
elif self.username and \
str2bool(config.get('container_auth_enabled', False)):
log.debug('Auth User lookup by USER NAME %s' % self.username)
dbuser = login_container_auth(self.username)
if dbuser is not None:
log.debug('filling all attributes to object')
for k, v in dbuser.get_dict().items():
setattr(self, k, v)
self.set_authenticated()
is_user_loaded = True
log.debug('No data in %s that could been used to log in' % self)
if not is_user_loaded:
# if we cannot authenticate user try anonymous
if self.anonymous_user.active is True:
user_model.fill_data(self, user_id=self.anonymous_user.user_id)
# then we set this user is logged in
self.is_authenticated = True
self.user_id = None
self.username = None
if not self.username:
self.username = 'None'
log.debug('Auth User is now %s' % self)
user_model.fill_perms(self)
@property
def is_admin(self):
return self.admin
def ip_allowed(self):
Checks if ip_addr used in constructor is allowed from defined list of
allowed ip_addresses for user
:returns: boolean, True if ip is in allowed ip range
#check IP
allowed_ips = AuthUser.get_allowed_ips(self.user_id, cache=True)
if check_ip_access(source_ip=self.ip_addr, allowed_ips=allowed_ips):
log.debug('IP:%s is in range of %s' % (self.ip_addr, allowed_ips))
log.info('Access for IP:%s forbidden, '
'not in %s' % (self.ip_addr, allowed_ips))
return "<AuthUser('id:%s:%s|%s')>" % (self.user_id, self.username,
self.is_authenticated)
def set_authenticated(self, authenticated=True):
if self.user_id != self.anonymous_user.user_id:
self.is_authenticated = authenticated
def get_cookie_store(self):
return {'username': self.username,
'user_id': self.user_id,
'is_authenticated': self.is_authenticated}
def from_cookie_store(cls, cookie_store):
Creates AuthUser from a cookie store
:param cls:
:param cookie_store:
user_id = cookie_store.get('user_id')
username = cookie_store.get('username')
api_key = cookie_store.get('api_key')
return AuthUser(user_id, api_key, username)
def get_allowed_ips(cls, user_id, cache=False):
_set = set()
user_ips = UserIpMap.query().filter(UserIpMap.user_id == user_id)
if cache:
user_ips = user_ips.options(FromCache("sql_cache_short",
"get_user_ips_%s" % user_id))
for ip in user_ips:
_set.add(ip.ip_addr)
except ObjectDeletedError:
# since we use heavy caching sometimes it happens that we get
# deleted objects here, we just skip them
return _set or set(['0.0.0.0/0', '::/0'])
def set_available_permissions(config):
This function will propagate pylons globals with all available defined
permission given in db. We don't want to check each time from db for new
permissions since adding a new permission also requires application restart
ie. to decorate new views with the newly created permission
:param config: current pylons config instance
log.info('getting information about all available permissions')
sa = meta.Session
all_perms = sa.query(Permission).all()
except Exception:
finally:
meta.Session.remove()
config['available_permissions'] = [x.permission_name for x in all_perms]
#==============================================================================
# CHECK DECORATORS
class LoginRequired(object):
Must be logged in to execute this function else
redirect to login page
:param api_access: if enabled this checks only for valid auth token
and grants access based on valid token
def __init__(self, api_access=False):
self.api_access = api_access
def __call__(self, func):
return decorator(self.__wrapper, func)
def __wrapper(self, func, *fargs, **fkwargs):
cls = fargs[0]
user = cls.rhodecode_user
loc = "%s:%s" % (cls.__class__.__name__, func.__name__)
ip_access_ok = True
if not user.ip_allowed:
from rhodecode.lib import helpers as h
h.flash(h.literal(_('IP %s not allowed' % (user.ip_addr))),
category='warning')
ip_access_ok = False
api_access_ok = False
if self.api_access:
log.debug('Checking API KEY access for %s' % cls)
if user.api_key == request.GET.get('api_key'):
api_access_ok = True
log.debug("API KEY token not valid")
log.debug('Checking if %s is authenticated @ %s' % (user.username, loc))
if (user.is_authenticated or api_access_ok) and ip_access_ok:
reason = 'RegularAuth' if user.is_authenticated else 'APIAuth'
log.info('user %s is authenticated and granted access to %s '
'using %s' % (user.username, loc, reason)
)
return func(*fargs, **fkwargs)
log.warn('user %s NOT authenticated on func: %s' % (
user, loc)
p = url.current()
log.debug('redirecting to login page with %s' % p)
return redirect(url('login_home', came_from=p))
class NotAnonymous(object):
redirect to login page"""
self.user = cls.rhodecode_user
log.debug('Checking if user is not anonymous @%s' % cls)
anonymous = self.user.username == 'default'
if anonymous:
h.flash(_('You need to be a registered user to '
'perform this action'),
class PermsDecorator(object):
"""Base class for controller decorators"""
def __init__(self, *required_perms):
available_perms = config['available_permissions']
for perm in required_perms:
if perm not in available_perms:
raise Exception("'%s' permission is not defined" % perm)
self.required_perms = set(required_perms)
self.user_perms = None
self.user_perms = self.user.permissions
log.debug('checking %s permissions %s for %s %s',
self.__class__.__name__, self.required_perms, cls, self.user)
if self.check_permissions():
log.debug('Permission granted for %s %s' % (cls, self.user))
log.debug('Permission denied for %s %s' % (cls, self.user))
h.flash(_('You need to be a signed in to '
'view this page'),
# redirect with forbidden ret code
return abort(403)
def check_permissions(self):
"""Dummy function for overriding"""
raise Exception('You have to write this function in child class')
class HasPermissionAllDecorator(PermsDecorator):
Checks for access permission for all given predicates. All of them
have to be meet in order to fulfill the request
if self.required_perms.issubset(self.user_perms.get('global')):
class HasPermissionAnyDecorator(PermsDecorator):
Checks for access permission for any of given predicates. In order to
fulfill the request any of predicates must be meet
if self.required_perms.intersection(self.user_perms.get('global')):
class HasRepoPermissionAllDecorator(PermsDecorator):
Checks for access permission for all given predicates for specific
repository. All of them have to be meet in order to fulfill the request
repo_name = get_repo_slug(request)
user_perms = set([self.user_perms['repositories'][repo_name]])
except KeyError:
if self.required_perms.issubset(user_perms):
class HasRepoPermissionAnyDecorator(PermsDecorator):
Checks for access permission for any of given predicates for specific
repository. In order to fulfill the request any of predicates must be meet
if self.required_perms.intersection(user_perms):
class HasReposGroupPermissionAllDecorator(PermsDecorator):
group_name = get_repos_group_slug(request)
user_perms = set([self.user_perms['repositories_groups'][group_name]])
class HasReposGroupPermissionAnyDecorator(PermsDecorator):
# CHECK FUNCTIONS
class PermsFunction(object):
"""Base function for other check functions"""
def __init__(self, *perms):
for perm in perms:
self.required_perms = set(perms)
self.repo_name = None
self.group_name = None
def __call__(self, check_location=''):
#TODO: put user as attribute here
user = request.user
cls_name = self.__class__.__name__
check_scope = {
'HasPermissionAll': '',
'HasPermissionAny': '',
'HasRepoPermissionAll': 'repo:%s' % self.repo_name,
'HasRepoPermissionAny': 'repo:%s' % self.repo_name,
'HasReposGroupPermissionAll': 'group:%s' % self.group_name,
'HasReposGroupPermissionAny': 'group:%s' % self.group_name,
}.get(cls_name, '?')
log.debug('checking cls:%s %s usr:%s %s @ %s', cls_name,
self.required_perms, user, check_scope,
check_location or 'unspecified location')
log.debug('Empty request user')
self.user_perms = user.permissions
log.debug('Permission to %s granted for user: %s @ %s', self.repo_name, user,
log.debug('Permission to %s denied for user: %s @ %s', self.repo_name, user,
class HasPermissionAll(PermsFunction):
class HasPermissionAny(PermsFunction):
class HasRepoPermissionAll(PermsFunction):
def __call__(self, repo_name=None, check_location=''):
self.repo_name = repo_name
return super(HasRepoPermissionAll, self).__call__(check_location)
if not self.repo_name:
self.repo_name = get_repo_slug(request)
self._user_perms = set(
[self.user_perms['repositories'][self.repo_name]]
if self.required_perms.issubset(self._user_perms):
class HasRepoPermissionAny(PermsFunction):
return super(HasRepoPermissionAny, self).__call__(check_location)
if self.required_perms.intersection(self._user_perms):
class HasReposGroupPermissionAny(PermsFunction):
def __call__(self, group_name=None, check_location=''):
self.group_name = group_name
return super(HasReposGroupPermissionAny, self).__call__(check_location)
[self.user_perms['repositories_groups'][self.group_name]]
class HasReposGroupPermissionAll(PermsFunction):
return super(HasReposGroupPermissionAll, self).__call__(check_location)
# SPECIAL VERSION TO HANDLE MIDDLEWARE AUTH
class HasPermissionAnyMiddleware(object):
def __call__(self, user, repo_name):
# repo_name MUST be unicode, since we handle keys in permission
# dict by unicode
repo_name = safe_unicode(repo_name)
usr = AuthUser(user.user_id)
self.user_perms = set([usr.permissions['repositories'][repo_name]])
log.error('Exception while accessing permissions %s' %
traceback.format_exc())
self.user_perms = set()
self.username = user.username
return self.check_permissions()
log.debug('checking VCS protocol '
'permissions %s for user:%s repository:%s', self.user_perms,
self.username, self.repo_name)
if self.required_perms.intersection(self.user_perms):
log.debug('permission granted for user:%s on repo:%s' % (
self.username, self.repo_name
log.debug('permission denied for user:%s on repo:%s' % (
# SPECIAL VERSION TO HANDLE API AUTH
class _BaseApiPerm(object):
def __call__(self, check_location='unspecified', user=None, repo_name=None):
check_scope = 'user:%s, repo:%s' % (user, repo_name)
log.debug('checking cls:%s %s %s @ %s', cls_name,
self.required_perms, check_scope, check_location)
log.debug('Empty User passed into arguments')
## process user
if not isinstance(user, AuthUser):
user = AuthUser(user.user_id)
if self.check_permissions(user.permissions, repo_name):
log.debug('Permission to %s granted for user: %s @ %s', repo_name,
user, check_location)
log.debug('Permission to %s denied for user: %s @ %s', repo_name,
def check_permissions(self, perm_defs, repo_name):
implement in child class should return True if permissions are ok,
False otherwise
:param perm_defs: dict with permission definitions
:param repo_name: repo name
raise NotImplementedError()
class HasPermissionAllApi(_BaseApiPerm):
def __call__(self, user, check_location=''):
return super(HasPermissionAllApi, self)\
.__call__(check_location=check_location, user=user)
def check_permissions(self, perm_defs, repo):
if self.required_perms.issubset(perm_defs.get('global')):
class HasPermissionAnyApi(_BaseApiPerm):
return super(HasPermissionAnyApi, self)\
if self.required_perms.intersection(perm_defs.get('global')):
class HasRepoPermissionAllApi(_BaseApiPerm):
def __call__(self, user, repo_name, check_location=''):
return super(HasRepoPermissionAllApi, self)\
.__call__(check_location=check_location, user=user,
repo_name=repo_name)
[perm_defs['repositories'][repo_name]]
log.warning(traceback.format_exc())
class HasRepoPermissionAnyApi(_BaseApiPerm):
return super(HasRepoPermissionAnyApi, self)\
_user_perms = set(
if self.required_perms.intersection(_user_perms):
def check_ip_access(source_ip, allowed_ips=None):
Checks if source_ip is a subnet of any of allowed_ips.
:param source_ip:
:param allowed_ips: list of allowed ips together with mask
from rhodecode.lib import ipaddr
log.debug('checking if ip:%s is subnet of %s' % (source_ip, allowed_ips))
if isinstance(allowed_ips, (tuple, list, set)):
for ip in allowed_ips:
if ipaddr.IPAddress(source_ip) in ipaddr.IPNetwork(ip):
# for any case we cannot determine the IP, don't crash just
# skip it and log as error, we want to say forbidden still when
# sending bad IP
continue
rhodecode.model.db
Database Models for RhodeCode
:created_on: Apr 08, 2010
import os
import time
from collections import defaultdict
from sqlalchemy import *
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import relationship, joinedload, class_mapper, validates
from sqlalchemy.exc import DatabaseError
from beaker.cache import cache_region, region_invalidate
from webob.exc import HTTPNotFound
from pylons.i18n.translation import lazy_ugettext as _
from rhodecode.lib.vcs import get_backend
from rhodecode.lib.vcs.utils.helpers import get_scm
from rhodecode.lib.vcs.exceptions import VCSError
from rhodecode.lib.vcs.utils.lazy import LazyProperty
from rhodecode.lib.utils2 import str2bool, safe_str, get_changeset_safe, \
safe_unicode, remove_suffix, remove_prefix
from rhodecode.lib.compat import json
from rhodecode.model.meta import Base, Session
URL_SEP = '/'
# BASE CLASSES
_hash_key = lambda k: hashlib.md5(safe_str(k)).hexdigest()
class BaseModel(object):
Base Model for all classess
def _get_keys(cls):
"""return column names for this model """
return class_mapper(cls).c.keys()
def get_dict(self):
return dict with keys and values corresponding
to this model data """
d = {}
for k in self._get_keys():
d[k] = getattr(self, k)
# also use __json__() if present to get additional fields
_json_attr = getattr(self, '__json__', None)
if _json_attr:
# update with attributes from __json__
if callable(_json_attr):
_json_attr = _json_attr()
for k, val in _json_attr.iteritems():
d[k] = val
return d
def get_appstruct(self):
"""return list with keys and values tupples corresponding
l = []
l.append((k, getattr(self, k),))
return l
def populate_obj(self, populate_dict):
"""populate model with data from given populate_dict"""
if k in populate_dict:
setattr(self, k, populate_dict[k])
def query(cls):
return Session().query(cls)
def get(cls, id_):
if id_:
return cls.query().get(id_)
def get_or_404(cls, id_):
id_ = int(id_)
except (TypeError, ValueError):
raise HTTPNotFound
res = cls.query().get(id_)
if not res:
return res
def getAll(cls):
return cls.query().all()
def delete(cls, id_):
obj = cls.query().get(id_)
Session().delete(obj)
if hasattr(self, '__unicode__'):
# python repr needs to return str
return safe_str(self.__unicode__())
return '<DB:%s>' % (self.__class__.__name__)
class RhodeCodeSetting(Base, BaseModel):
__tablename__ = 'rhodecode_settings'
__table_args__ = (
UniqueConstraint('app_settings_name'),
{'extend_existing': True, 'mysql_engine': 'InnoDB',
'mysql_charset': 'utf8'}
app_settings_id = Column("app_settings_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
app_settings_name = Column("app_settings_name", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
_app_settings_value = Column("app_settings_value", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
def __init__(self, k='', v=''):
self.app_settings_name = k
self.app_settings_value = v
@validates('_app_settings_value')
def validate_settings_value(self, key, val):
assert type(val) == unicode
return val
@hybrid_property
def app_settings_value(self):
v = self._app_settings_value
if self.app_settings_name in ["ldap_active",
"default_repo_enable_statistics",
"default_repo_enable_locking",
"default_repo_private",
"default_repo_enable_downloads"]:
v = str2bool(v)
return v
@app_settings_value.setter
def app_settings_value(self, val):
Setter that will always make sure we use unicode in app_settings_value
:param val:
self._app_settings_value = safe_unicode(val)
def __unicode__(self):
return u"<%s('%s:%s')>" % (
self.__class__.__name__,
self.app_settings_name, self.app_settings_value
def get_by_name(cls, key):
return cls.query()\
.filter(cls.app_settings_name == key).scalar()
def get_by_name_or_create(cls, key):
res = cls.get_by_name(key)
res = cls(key)
def get_app_settings(cls, cache=False):
ret = cls.query()
ret = ret.options(FromCache("sql_cache_short", "get_hg_settings"))
if not ret:
raise Exception('Could not get application settings !')
settings = {}
for each in ret:
settings['rhodecode_' + each.app_settings_name] = \
each.app_settings_value
return settings
def get_ldap_settings(cls, cache=False):
ret = cls.query()\
.filter(cls.app_settings_name.startswith('ldap_')).all()
fd = {}
for row in ret:
fd.update({row.app_settings_name: row.app_settings_value})
return fd
def get_default_repo_settings(cls, cache=False, strip_prefix=False):
.filter(cls.app_settings_name.startswith('default_')).all()
key = row.app_settings_name
if strip_prefix:
key = remove_prefix(key, prefix='default_')
fd.update({key: row.app_settings_value})
class RhodeCodeUi(Base, BaseModel):
__tablename__ = 'rhodecode_ui'
UniqueConstraint('ui_key'),
HOOK_UPDATE = 'changegroup.update'
HOOK_REPO_SIZE = 'changegroup.repo_size'
HOOK_PUSH = 'changegroup.push_logger'
HOOK_PRE_PUSH = 'prechangegroup.pre_push'
HOOK_PULL = 'outgoing.pull_logger'
HOOK_PRE_PULL = 'preoutgoing.pre_pull'
ui_id = Column("ui_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
ui_section = Column("ui_section", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
ui_key = Column("ui_key", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
ui_value = Column("ui_value", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
ui_active = Column("ui_active", Boolean(), nullable=True, unique=None, default=True)
def get_by_key(cls, key):
return cls.query().filter(cls.ui_key == key).scalar()
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]))
return q.all()
def get_custom_hooks(cls):
q = q.filter(~cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE,
q = q.filter(cls.ui_section == 'hooks')
def get_repos_location(cls):
return cls.get_by_key('/').ui_value
def create_or_update_hook(cls, key, val):
new_ui = cls.get_by_key(key) or cls()
new_ui.ui_section = 'hooks'
new_ui.ui_active = True
new_ui.ui_key = key
new_ui.ui_value = val
Session().add(new_ui)
return '<DB:%s[%s:%s]>' % (self.__class__.__name__, self.ui_key,
self.ui_value)
class User(Base, BaseModel):
__tablename__ = 'users'
UniqueConstraint('username'), UniqueConstraint('email'),
Index('u_username_idx', 'username'),
Index('u_email_idx', 'email'),
DEFAULT_USER = 'default'
DEFAULT_PERMISSIONS = [
'hg.register.manual_activate', 'hg.create.repository',
'hg.fork.repository', 'repository.read', 'group.read'
]
user_id = Column("user_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
username = Column("username", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
password = Column("password", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
active = Column("active", Boolean(), nullable=True, unique=None, default=True)
admin = Column("admin", Boolean(), nullable=True, unique=None, default=False)
name = Column("firstname", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
lastname = Column("lastname", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
_email = Column("email", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
last_login = Column("last_login", DateTime(timezone=False), nullable=True, unique=None, default=None)
ldap_dn = Column("ldap_dn", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
api_key = Column("api_key", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
inherit_default_permissions = Column("inherit_default_permissions", Boolean(), nullable=False, unique=None, default=True)
user_log = relationship('UserLog')
user_perms = relationship('UserToPerm', primaryjoin="User.user_id==UserToPerm.user_id", cascade='all')
repositories = relationship('Repository')
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('UsersGroupMember', 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')
def email(self):
return self._email
@email.setter
def email(self, val):
self._email = val.lower() if val else None
def firstname(self):
# alias for future
return self.name
def emails(self):
other = UserEmailMap.query().filter(UserEmailMap.user==self).all()
return [self.email] + [x.email for x in other]
def ip_addresses(self):
ret = UserIpMap.query().filter(UserIpMap.user == self).all()
return [x.ip_addr for x in ret]
def username_and_name(self):
return '%s (%s %s)' % (self.username, self.firstname, self.lastname)
def full_name(self):
return '%s %s' % (self.firstname, self.lastname)
def full_name_or_username(self):
return ('%s %s' % (self.firstname, self.lastname)
if (self.firstname and self.lastname) else self.username)
def full_contact(self):
return '%s %s <%s>' % (self.firstname, self.lastname, self.email)
def short_contact(self):
def AuthUser(self):
Returns instance of AuthUser for this user
from rhodecode.lib.auth import AuthUser
return AuthUser(user_id=self.user_id, api_key=self.api_key,
username=self.username)
return u"<%s('id:%s:%s')>" % (self.__class__.__name__,
self.user_id, self.username)
def get_by_username(cls, username, case_insensitive=False, cache=False):
if case_insensitive:
q = cls.query().filter(cls.username.ilike(username))
q = cls.query().filter(cls.username == username)
q = q.options(FromCache(
"sql_cache_short",
"get_user_%s" % _hash_key(username)
return q.scalar()
def get_by_api_key(cls, api_key, cache=False):
q = cls.query().filter(cls.api_key == api_key)
q = q.options(FromCache("sql_cache_short",
"get_api_key_%s" % api_key))
def get_by_email(cls, email, case_insensitive=False, cache=False):
q = cls.query().filter(cls.email.ilike(email))
q = cls.query().filter(cls.email == email)
"get_email_key_%s" % email))
ret = q.scalar()
if ret is None:
q = UserEmailMap.query()
# try fetching in alternate email map
q = q.filter(UserEmailMap.email.ilike(email))
q = q.filter(UserEmailMap.email == email)
q = q.options(joinedload(UserEmailMap.user))
"get_email_map_key_%s" % email))
ret = getattr(q.scalar(), 'user', None)
return ret
def get_from_cs_author(cls, author):
Tries to get User objects out of commit author string
:param author:
from rhodecode.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, case_insensitive=True)
if user:
# Maybe we can match by username?
_author = author_name(author)
user = cls.get_by_username(_author, case_insensitive=True)
def update_lastlogin(self):
"""Update user lastlogin"""
self.last_login = datetime.datetime.now()
Session().add(self)
log.debug('updated user %s lastlogin' % self.username)
def get_api_data(self):
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,
api_key=user.api_key,
active=user.active,
admin=user.admin,
ldap_dn=user.ldap_dn,
last_login=user.last_login,
ip_addresses=user.ip_addresses
return data
def __json__(self):
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())
class UserEmailMap(Base, BaseModel):
__tablename__ = 'user_email_map'
Index('uem_email_idx', 'email'),
UniqueConstraint('email'),
__mapper_args__ = {}
email_id = Column("email_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=True, unique=None, default=None)
_email = Column("email", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=False, default=None)
user = relationship('User', lazy='joined')
@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
class UserIpMap(Base, BaseModel):
__tablename__ = 'user_ip_map'
UniqueConstraint('user_id', 'ip_addr'),
ip_id = Column("ip_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
ip_addr = Column("ip_addr", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=False, default=None)
def _get_ip_range(cls, ip_addr):
net = ipaddr.IPNetwork(address=ip_addr)
return [str(net.network), str(net.broadcast)]
return dict(
ip_addr=self.ip_addr,
ip_range=self._get_ip_range(self.ip_addr)
class UserLog(Base, BaseModel):
__tablename__ = 'user_logs'
'mysql_charset': 'utf8'},
user_log_id = Column("user_log_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
repository_id = Column("repository_id", Integer(), ForeignKey('repositories.repo_id'), nullable=True)
repository_name = Column("repository_name", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
user_ip = Column("user_ip", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
action = Column("action", UnicodeText(1200000, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
action_date = Column("action_date", DateTime(timezone=False), nullable=True, unique=None, default=None)
def action_as_day(self):
return datetime.date(*self.action_date.timetuple()[:3])
user = relationship('User')
repository = relationship('Repository', cascade='')
class UsersGroup(Base, BaseModel):
__tablename__ = 'users_groups'
users_group_id = Column("users_group_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
users_group_name = Column("users_group_name", String(255, convert_unicode=False, assert_unicode=None), nullable=False, unique=True, default=None)
users_group_active = Column("users_group_active", Boolean(), nullable=True, unique=None, default=None)
inherit_default_permissions = Column("users_group_inherit_default_permissions", Boolean(), nullable=False, unique=None, default=True)
members = relationship('UsersGroupMember', cascade="all, delete, delete-orphan", lazy="joined")
users_group_to_perm = relationship('UsersGroupToPerm', cascade='all')
users_group_repo_to_perm = relationship('UsersGroupRepoToPerm', cascade='all')
return u'<userGroup(%s)>' % (self.users_group_name)
def get_by_group_name(cls, group_name, cache=False,
case_insensitive=False):
q = cls.query().filter(cls.users_group_name.ilike(group_name))
q = cls.query().filter(cls.users_group_name == group_name)
"get_user_%s" % _hash_key(group_name)
def get(cls, users_group_id, cache=False):
users_group = cls.query()
users_group = users_group.options(FromCache("sql_cache_short",
"get_users_group_%s" % users_group_id))
return users_group.get(users_group_id)
users_group = self
users_group_id=users_group.users_group_id,
group_name=users_group.users_group_name,
active=users_group.users_group_active,
class UsersGroupMember(Base, BaseModel):
__tablename__ = 'users_groups_members'
users_group_member_id = Column("users_group_member_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
users_group_id = Column("users_group_id", Integer(), ForeignKey('users_groups.users_group_id'), nullable=False, unique=None, default=None)
user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=False, unique=None, default=None)
users_group = relationship('UsersGroup')
def __init__(self, gr_id='', u_id=''):
self.users_group_id = gr_id
self.user_id = u_id
class RepositoryField(Base, BaseModel):
__tablename__ = 'repositories_fields'
UniqueConstraint('repository_id', 'field_key'), # no-multi field
PREFIX = 'ex_' # prefix used in form to not conflict with already existing fields
repo_field_id = Column("repo_field_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
repository_id = Column("repository_id", Integer(), ForeignKey('repositories.repo_id'), nullable=False, unique=None, default=None)
field_key = Column("field_key", String(250, convert_unicode=False, assert_unicode=None))
field_label = Column("field_label", String(1024, convert_unicode=False, assert_unicode=None), nullable=False)
field_value = Column("field_value", String(10000, convert_unicode=False, assert_unicode=None), nullable=False)
field_desc = Column("field_desc", String(1024, convert_unicode=False, assert_unicode=None), nullable=False)
field_type = Column("field_type", String(256), nullable=False, unique=None)
created_on = Column('created_on', DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
repository = relationship('Repository')
def field_key_prefixed(self):
return 'ex_%s' % self.field_key
def un_prefix_key(cls, key):
if key.startswith(cls.PREFIX):
return key[len(cls.PREFIX):]
return key
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, BaseModel):
__tablename__ = 'repositories'
UniqueConstraint('repo_name'),
Index('r_repo_name_idx', 'repo_name'),
repo_id = Column("repo_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
repo_name = Column("repo_name", String(255, convert_unicode=False, assert_unicode=None), nullable=False, unique=True, default=None)
clone_uri = Column("clone_uri", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=False, default=None)
repo_type = Column("repo_type", String(255, convert_unicode=False, assert_unicode=None), nullable=False, unique=False, default=None)
user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=False, unique=False, default=None)
private = Column("private", Boolean(), nullable=True, unique=None, default=None)
enable_statistics = Column("statistics", Boolean(), nullable=True, unique=None, default=True)
enable_downloads = Column("downloads", Boolean(), nullable=True, unique=None, default=True)
description = Column("description", String(10000, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
created_on = Column('created_on', DateTime(timezone=False), nullable=True, unique=None, default=datetime.datetime.now)
updated_on = Column('updated_on', DateTime(timezone=False), nullable=True, unique=None, default=datetime.datetime.now)
landing_rev = Column("landing_revision", String(255, convert_unicode=False, assert_unicode=None), nullable=False, unique=False, default=None)
enable_locking = Column("enable_locking", Boolean(), nullable=False, unique=None, default=False)
_locked = Column("locked", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=False, default=None)
_changeset_cache = Column("changeset_cache", LargeBinary(), nullable=True) #JSON data
fork_id = Column("fork_id", Integer(), ForeignKey('repositories.repo_id'), nullable=True, unique=False, default=None)
group_id = Column("group_id", Integer(), ForeignKey('groups.group_id'), nullable=True, unique=False, default=None)
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('UsersGroupRepoToPerm', cascade='all')
stats = relationship('Statistics', cascade='all', uselist=False)
followers = relationship('UserFollowing',
primaryjoin='UserFollowing.follows_repo_id==Repository.repo_id',
cascade='all')
extra_fields = relationship('RepositoryField',
cascade="all, delete, delete-orphan")
logs = relationship('UserLog')
comments = relationship('ChangesetComment', cascade="all, delete, delete-orphan")
pull_requests_org = relationship('PullRequest',
primaryjoin='PullRequest.org_repo_id==Repository.repo_id',
pull_requests_other = relationship('PullRequest',
primaryjoin='PullRequest.other_repo_id==Repository.repo_id',
return u"<%s('%s:%s')>" % (self.__class__.__name__, self.repo_id,
self.repo_name)
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))
self._locked = None
def changeset_cache(self):
from rhodecode.lib.vcs.backends.base import EmptyChangeset
dummy = EmptyChangeset().__json__()
if not self._changeset_cache:
return dummy
return json.loads(self._changeset_cache)
except TypeError:
@changeset_cache.setter
def changeset_cache(self, val):
self._changeset_cache = json.dumps(val)
except:
def url_sep(cls):
return URL_SEP
def normalize_repo_name(cls, repo_name):
Normalizes os specific repo_name to the format internally stored inside
dabatabase using URL_SEP
:param repo_name:
return cls.url_sep().join(repo_name.split(os.sep))
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.user))\
.options(joinedload(Repository.group))
def get_by_full_path(cls, repo_full_path):
repo_name = repo_full_path.split(cls.base_path(), 1)[-1]
repo_name = cls.normalize_repo_name(repo_name)
return cls.get_by_repo_name(repo_name.strip(URL_SEP))
def get_repo_forks(cls, repo_id):
return cls.query().filter(Repository.fork_id == repo_id)
def base_path(cls):
Returns base path when all repos are stored
q = Session().query(RhodeCodeUi)\
.filter(RhodeCodeUi.ui_key == cls.url_sep())
q = q.options(FromCache("sql_cache_short", "repository_repo_path"))
return q.one().ui_value
def forks(self):
Return forks of this repo
return Repository.get_repo_forks(self.repo_id)
def parent(self):
Returns fork parent
return self.fork
def just_name(self):
return self.repo_name.split(Repository.url_sep())[-1]
def groups_with_parents(self):
groups = []
if self.group is None:
return groups
cur_gr = self.group
groups.insert(0, cur_gr)
while 1:
gr = getattr(cur_gr, 'parent_group', None)
cur_gr = cur_gr.parent_group
if gr is None:
groups.insert(0, gr)
def groups_and_repo(self):
return self.groups_with_parents, self.just_name
@LazyProperty
def repo_path(self):
Returns base full path for that repository means where it actually
exists on a filesystem
q = Session().query(RhodeCodeUi).filter(RhodeCodeUi.ui_key ==
Repository.url_sep())
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(*p)
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])
def _ui(self):
Creates an db based ui object for this repository
from rhodecode.lib.utils import make_ui
return make_ui('db', clear_session=False)
def inject_ui(cls, repo, extras={}):
from rhodecode.lib.vcs.backends.hg import MercurialRepository
from rhodecode.lib.vcs.backends.git import GitRepository
required = (MercurialRepository, GitRepository)
if not isinstance(repo, required):
raise Exception('repo must be instance of %s' % required)
# inject ui extra param to log this action via push logger
for k, v in extras.items():
repo._repo.ui.setconfig('rhodecode_extras', k, v)
def is_valid(cls, repo_name):
returns True if given repo name is a valid filesystem repository
from rhodecode.lib.utils import is_valid_repo
return is_valid_repo(repo_name, cls.base_path())
Common function for generating repo api data
repo = self
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.user.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
rc_config = RhodeCodeSetting.get_app_settings()
repository_fields = str2bool(rc_config.get('rhodecode_repository_fields'))
if repository_fields:
for f in self.extra_fields:
data[f.field_key_prefixed] = f.field_value
def lock(cls, repo, user_id):
repo.locked = [user_id, time.time()]
Session().add(repo)
def unlock(cls, repo):
repo.locked = None
def last_db_change(self):
return self.updated_on
def clone_url(self, **override):
from pylons import url
from urlparse import urlparse
import urllib
parsed_url = urlparse(url('home', qualified=True))
default_clone_uri = '%(scheme)s://%(user)s%(pass)s%(netloc)s%(prefix)s%(path)s'
decoded_path = safe_unicode(urllib.unquote(parsed_url.path))
args = {
'user': '',
'pass': '',
'scheme': parsed_url.scheme,
'netloc': parsed_url.netloc,
'prefix': decoded_path,
'path': self.repo_name
args.update(override)
return default_clone_uri % args
#==========================================================================
# SCM PROPERTIES
def get_changeset(self, rev=None):
return get_changeset_safe(self.scm_instance, rev)
def get_landing_changeset(self):
Returns landing changeset, or if that doesn't exist returns the tip
cs = self.get_changeset(self.landing_rev) or self.get_changeset()
return cs
def update_changeset_cache(self, cs_cache=None):
Update cache of last changeset for repository, keys should be::
short_id
raw_id
revision
message
date
author
:param cs_cache:
from rhodecode.lib.vcs.backends.base import BaseChangeset
if cs_cache is None:
cs_cache = self.get_changeset()
if isinstance(cs_cache, BaseChangeset):
cs_cache = cs_cache.__json__()
if (cs_cache != self.changeset_cache
or not self.last_change
or not self.changeset_cache):
_default = datetime.datetime.fromtimestamp(0)
last_change = cs_cache.get('date') or self.last_change or _default
log.debug('updated repo %s with new cs cache %s' % (self, cs_cache))
self.updated_on = last_change
self.changeset_cache = cs_cache
log.debug('Skipping repo:%s already with latest changes' % self)
def tip(self):
return self.get_changeset('tip')
def author(self):
return self.tip.author
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:
cmts = cmts.filter(ChangesetComment.revision.in_(revisions))
grouped = defaultdict(list)
for cmt in cmts.all():
grouped[cmt.revision].append(cmt)
return grouped
def statuses(self, revisions=None):
Returns statuses for this repository
:param revisions: list of revisions to get statuses for
:type revisions: list
statuses = ChangesetStatus.query()\
.filter(ChangesetStatus.repo == self)\
.filter(ChangesetStatus.version == 0)
statuses = statuses.filter(ChangesetStatus.revision.in_(revisions))
grouped = {}
#maybe we have open new pullrequest without a status ?
stat = ChangesetStatus.STATUS_UNDER_REVIEW
status_lbl = ChangesetStatus.get_status_lbl(stat)
for pr in PullRequest.query().filter(PullRequest.org_repo == self).all():
for rev in pr.revisions:
pr_id = pr.pull_request_id
pr_repo = pr.other_repo.repo_name
grouped[rev] = [stat, status_lbl, pr_id, pr_repo]
for stat in statuses.all():
pr_id = pr_repo = None
if stat.pull_request:
pr_id = stat.pull_request.pull_request_id
pr_repo = stat.pull_request.other_repo.repo_name
grouped[stat.revision] = [str(stat.status), stat.status_lbl,
pr_id, pr_repo]
def _repo_size(self):
log.debug('calculating repository size...')
return h.format_byte_size(self.scm_instance.size)
# SCM CACHE INSTANCE
def invalidate(self):
return CacheInvalidation.invalidate(self.repo_name)
def set_invalidate(self):
set a cache for invalidation for this instance
CacheInvalidation.set_invalidate(repo_name=self.repo_name)
def scm_instance_no_cache(self):
return self.__get_instance()
def scm_instance(self):
import rhodecode
full_cache = str2bool(rhodecode.CONFIG.get('vcs_full_cache'))
if full_cache:
return self.scm_instance_cached()
def scm_instance_cached(self, cache_map=None):
@cache_region('long_term')
def _c(repo_name):
rn = self.repo_name
log.debug('Getting cached instance of repo')
if cache_map:
# get using prefilled cache_map
invalidate_repo = cache_map[self.repo_name]
if invalidate_repo:
invalidate_repo = (None if invalidate_repo.cache_active
else invalidate_repo)
# get from invalidate
invalidate_repo = self.invalidate
if invalidate_repo is not None:
region_invalidate(_c, None, rn)
# update our cache
CacheInvalidation.set_valid(invalidate_repo.cache_key)
return _c(rn)
def __get_instance(self):
repo_full_path = self.repo_full_path
alias = get_scm(repo_full_path)[0]
log.debug('Creating instance of %s repository' % alias)
backend = get_backend(alias)
except VCSError:
log.error('Perhaps this repository is in db and not in '
'filesystem run rescan repositories with '
'"destroy old data " option from admin panel')
return
if alias == 'hg':
repo = backend(safe_str(repo_full_path), create=False,
baseui=self._ui)
# skip hidden web repository
if repo._get_hidden():
repo = backend(repo_full_path, create=False)
return repo
class RepoGroup(Base, BaseModel):
__tablename__ = 'groups'
UniqueConstraint('group_name', 'group_parent_id'),
CheckConstraint('group_id != group_parent_id'),
__mapper_args__ = {'order_by': 'group_name'}
group_id = Column("group_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
group_name = Column("group_name", String(255, convert_unicode=False, assert_unicode=None), nullable=False, unique=True, default=None)
group_parent_id = Column("group_parent_id", Integer(), ForeignKey('groups.group_id'), nullable=True, unique=None, default=None)
group_description = Column("group_description", String(10000, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
repo_group_to_perm = relationship('UserRepoGroupToPerm', cascade='all', order_by='UserRepoGroupToPerm.group_to_perm_id')
users_group_to_perm = relationship('UsersGroupRepoGroupToPerm', cascade='all')
parent_group = relationship('RepoGroup', remote_side=group_id)
def __init__(self, group_name='', parent_group=None):
self.parent_group = parent_group
return u"<%s('%s:%s')>" % (self.__class__.__name__, self.group_id,
self.group_name)
def groups_choices(cls, groups=None, show_empty_group=True):
from webhelpers.html import literal as _literal
if not groups:
groups = cls.query().all()
repo_groups = []
if show_empty_group:
repo_groups = [('-1', '-- no parent --')]
sep = ' » '
_name = lambda k: _literal(sep.join(k))
repo_groups.extend([(x.group_id, _name(x.full_path_splitted))
for x in groups])
repo_groups = sorted(repo_groups, key=lambda t: t[1].split(sep)[0])
return repo_groups
def get_by_group_name(cls, group_name, cache=False, case_insensitive=False):
gr = cls.query()\
.filter(cls.group_name.ilike(group_name))
.filter(cls.group_name == group_name)
gr = gr.options(FromCache(
"get_group_%s" % _hash_key(group_name)
return gr.scalar()
def parents(self):
parents_recursion_limit = 5
if self.parent_group is None:
cur_gr = self.parent_group
cnt = 0
cnt += 1
if cnt == parents_recursion_limit:
# this will prevent accidental infinit loops
log.error('group nested more than %s' %
parents_recursion_limit)
def children(self):
return RepoGroup.query().filter(RepoGroup.parent_group == self)
def name(self):
return self.group_name.split(RepoGroup.url_sep())[-1]
def full_path(self):
return self.group_name
def full_path_splitted(self):
return self.group_name.split(RepoGroup.url_sep())
def repositories(self):
return Repository.query()\
.filter(Repository.group == self)\
.order_by(Repository.repo_name)
def repositories_recursive_count(self):
cnt = self.repositories.count()
def children_count(group):
for child in group.children:
cnt += child.repositories.count()
cnt += children_count(child)
return cnt
return cnt + children_count(self)
def recursive_groups_and_repos(self):
Recursive return all groups, with repositories in those groups
all_ = []
def _get_members(root_gr):
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 get_new_name(self, group_name):
returns new full group name based on parent and new name
path_prefix = (self.parent_group.full_path_splitted if
self.parent_group else [])
return RepoGroup.url_sep().join(path_prefix + [group_name])
class Permission(Base, BaseModel):
__tablename__ = 'permissions'
Index('p_perm_name_idx', 'permission_name'),
PERMS = [
('repository.none', _('Repository no access')),
('repository.read', _('Repository read access')),
('repository.write', _('Repository write access')),
('repository.admin', _('Repository admin access')),
('group.none', _('Repositories Group no access')),
('group.read', _('Repositories Group read access')),
('group.write', _('Repositories Group write access')),
('group.admin', _('Repositories Group admin access')),
('hg.admin', _('RhodeCode Administrator')),
('hg.create.none', _('Repository creation disabled')),
('hg.create.repository', _('Repository creation enabled')),
('hg.fork.none', _('Repository forking disabled')),
('hg.fork.repository', _('Repository forking enabled')),
('hg.register.none', _('Register disabled')),
('hg.register.manual_activate', _('Register new user with RhodeCode '
'with manual activation')),
('hg.register.auto_activate', _('Register new user with RhodeCode '
'with auto activation')),
# defines which permissions are more important higher 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,
'hg.fork.none': 0,
'hg.fork.repository': 1,
'hg.create.none': 0,
'hg.create.repository':1
permission_id = Column("permission_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
permission_name = Column("permission_name", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
permission_longname = Column("permission_longname", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
self.__class__.__name__, self.permission_id, self.permission_name
return cls.query().filter(cls.permission_name == key).scalar()
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)
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)
class UserRepoToPerm(Base, BaseModel):
__tablename__ = 'repo_to_perm'
UniqueConstraint('user_id', 'repository_id', 'permission_id'),
repo_to_perm_id = Column("repo_to_perm_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
permission_id = Column("permission_id", Integer(), ForeignKey('permissions.permission_id'), nullable=False, unique=None, default=None)
permission = relationship('Permission')
def create(cls, user, repository, permission):
n = cls()
n.user = user
n.repository = repository
n.permission = permission
Session().add(n)
return n
return u'<user:%s => %s >' % (self.user, self.repository)
class UserToPerm(Base, BaseModel):
__tablename__ = 'user_to_perm'
UniqueConstraint('user_id', 'permission_id'),
user_to_perm_id = Column("user_to_perm_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
permission = relationship('Permission', lazy='joined')
class UsersGroupRepoToPerm(Base, BaseModel):
__tablename__ = 'users_group_repo_to_perm'
UniqueConstraint('repository_id', 'users_group_id', 'permission_id'),
users_group_to_perm_id = Column("users_group_to_perm_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
def create(cls, users_group, repository, permission):
n.users_group = users_group
return u'<userGroup:%s => %s >' % (self.users_group, self.repository)
class UsersGroupToPerm(Base, BaseModel):
__tablename__ = 'users_group_to_perm'
UniqueConstraint('users_group_id', 'permission_id',),
class UserRepoGroupToPerm(Base, BaseModel):
__tablename__ = 'user_repo_group_to_perm'
UniqueConstraint('user_id', 'group_id', 'permission_id'),
group_to_perm_id = Column("group_to_perm_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
group_id = Column("group_id", Integer(), ForeignKey('groups.group_id'), nullable=False, unique=None, default=None)
class UsersGroupRepoGroupToPerm(Base, BaseModel):
__tablename__ = 'users_group_repo_group_to_perm'
UniqueConstraint('users_group_id', 'group_id'),
users_group_repo_group_to_perm_id = Column("users_group_repo_group_to_perm_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
class Statistics(Base, BaseModel):
__tablename__ = 'statistics'
UniqueConstraint('repository_id'),
stat_id = Column("stat_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
repository_id = Column("repository_id", Integer(), ForeignKey('repositories.repo_id'), nullable=False, unique=True, default=None)
stat_on_revision = Column("stat_on_revision", Integer(), nullable=False)
commit_activity = Column("commit_activity", LargeBinary(1000000), nullable=False)#JSON data
commit_activity_combined = Column("commit_activity_combined", LargeBinary(), nullable=False)#JSON data
languages = Column("languages", LargeBinary(1000000), nullable=False)#JSON data
repository = relationship('Repository', single_parent=True)
class UserFollowing(Base, BaseModel):
__tablename__ = 'user_followings'
UniqueConstraint('user_id', 'follows_repository_id'),
UniqueConstraint('user_id', 'follows_user_id'),
user_following_id = Column("user_following_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
follows_repo_id = Column("follows_repository_id", Integer(), ForeignKey('repositories.repo_id'), nullable=True, unique=None, default=None)
follows_user_id = Column("follows_user_id", Integer(), ForeignKey('users.user_id'), nullable=True, unique=None, default=None)
follows_from = Column('follows_from', DateTime(timezone=False), nullable=True, unique=None, default=datetime.datetime.now)
user = relationship('User', primaryjoin='User.user_id==UserFollowing.user_id')
follows_user = relationship('User', primaryjoin='User.user_id==UserFollowing.follows_user_id')
follows_repository = relationship('Repository', order_by='Repository.repo_name')
def get_repo_followers(cls, repo_id):
return cls.query().filter(cls.follows_repo_id == repo_id)
class CacheInvalidation(Base, BaseModel):
__tablename__ = 'cache_invalidation'
UniqueConstraint('cache_key'),
Index('key_idx', 'cache_key'),
cache_id = Column("cache_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
cache_key = Column("cache_key", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
cache_args = Column("cache_args", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
cache_active = Column("cache_active", Boolean(), nullable=True, unique=None, default=False)
def __init__(self, cache_key, cache_args=''):
self.cache_key = cache_key
self.cache_args = cache_args
self.cache_active = False
return u"<%s('%s:%s')>" % (self.__class__.__name__,
self.cache_id, self.cache_key)
def prefix(self):
_split = self.cache_key.split(self.cache_args, 1)
if _split and len(_split) == 2:
return _split[0]
return ''
def clear_cache(cls):
cls.query().delete()
def _get_key(cls, key):
Wrapper for generating a key, together with a prefix
:param key:
prefix = ''
org_key = key
iid = rhodecode.CONFIG.get('instance_id')
if iid:
prefix = iid
return "%s%s" % (prefix, key), prefix, org_key
return cls.query().filter(cls.cache_key == key).scalar()
return cls.query().filter(cls.cache_args == repo_name).all()
def _get_or_create_key(cls, key, repo_name, commit=True):
inv_obj = Session().query(cls).filter(cls.cache_key == key).scalar()
if not inv_obj:
inv_obj = CacheInvalidation(key, repo_name)
Session().add(inv_obj)
if commit:
Session().rollback()
return inv_obj
def invalidate(cls, key):
Returns Invalidation object if this given key should be invalidated
None otherwise. `cache_active = False` means that this cache
state is not valid and needs to be invalidated
repo_name = key
repo_name = remove_suffix(repo_name, '_README')
repo_name = remove_suffix(repo_name, '_RSS')
repo_name = remove_suffix(repo_name, '_ATOM')
# adds instance prefix
key, _prefix, _org_key = cls._get_key(key)
inv = cls._get_or_create_key(key, repo_name)
if inv and inv.cache_active is False:
return inv
def set_invalidate(cls, key=None, repo_name=None):
Mark this Cache key for invalidation, either by key or whole
cache sets based on repo_name
invalidated_keys = []
if key:
inv_objs = Session().query(cls).filter(cls.cache_key == key).all()
elif repo_name:
inv_objs = Session().query(cls).filter(cls.cache_args == repo_name).all()
for inv_obj in inv_objs:
inv_obj.cache_active = False
log.debug('marking %s key for invalidation based on key=%s,repo_name=%s'
% (inv_obj, key, repo_name))
invalidated_keys.append(inv_obj.cache_key)
return invalidated_keys
def set_valid(cls, key):
Mark this cache key as active and currently cached
inv_obj = cls.get_by_key(key)
inv_obj.cache_active = True
def get_cache_map(cls):
class cachemapdict(dict):
def __init__(self, *args, **kwargs):
fixkey = kwargs.get('fixkey')
if fixkey:
del kwargs['fixkey']
self.fixkey = fixkey
super(cachemapdict, self).__init__(*args, **kwargs)
def __getattr__(self, name):
key = name
if self.fixkey:
if key in self.__dict__:
return self.__dict__[key]
return self[key]
def __getitem__(self, key):
return super(cachemapdict, self).__getitem__(key)
cache_map = cachemapdict(fixkey=True)
for obj in cls.query().all():
cache_map[obj.cache_key] = cachemapdict(obj.get_dict())
return cache_map
class ChangesetComment(Base, BaseModel):
__tablename__ = 'changeset_comments'
Index('cc_revision_idx', 'revision'),
comment_id = Column('comment_id', Integer(), nullable=False, primary_key=True)
repo_id = Column('repo_id', Integer(), ForeignKey('repositories.repo_id'), nullable=False)
revision = Column('revision', String(40), nullable=True)
pull_request_id = Column("pull_request_id", Integer(), ForeignKey('pull_requests.pull_request_id'), nullable=True)
line_no = Column('line_no', Unicode(10), nullable=True)
hl_lines = Column('hl_lines', Unicode(512), nullable=True)
f_path = Column('f_path', Unicode(1000), nullable=True)
user_id = Column('user_id', Integer(), ForeignKey('users.user_id'), nullable=False)
text = Column('text', UnicodeText(25000), nullable=False)
modified_at = Column('modified_at', DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
author = relationship('User', lazy='joined')
repo = relationship('Repository')
status_change = relationship('ChangesetStatus', cascade="all, delete, delete-orphan")
pull_request = relationship('PullRequest', lazy='joined')
def get_users(cls, revision=None, pull_request_id=None):
Returns user associated with this ChangesetComment. ie those
who actually commented
:param revision:
q = Session().query(User)\
.join(ChangesetComment.author)
if revision:
q = q.filter(cls.revision == revision)
elif pull_request_id:
q = q.filter(cls.pull_request_id == pull_request_id)
class ChangesetStatus(Base, BaseModel):
__tablename__ = 'changeset_statuses'
Index('cs_revision_idx', 'revision'),
Index('cs_version_idx', 'version'),
UniqueConstraint('repo_id', 'revision', 'version'),
STATUS_NOT_REVIEWED = DEFAULT = 'not_reviewed'
STATUS_APPROVED = 'approved'
STATUS_REJECTED = 'rejected'
STATUS_UNDER_REVIEW = 'under_review'
STATUSES = [
(STATUS_NOT_REVIEWED, _("Not Reviewed")), # (no icon) and default
(STATUS_APPROVED, _("Approved")),
(STATUS_REJECTED, _("Rejected")),
(STATUS_UNDER_REVIEW, _("Under Review")),
changeset_status_id = Column('changeset_status_id', Integer(), nullable=False, primary_key=True)
user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=False, unique=None)
revision = Column('revision', String(40), nullable=False)
status = Column('status', String(128), nullable=False, default=DEFAULT)
changeset_comment_id = Column('changeset_comment_id', Integer(), ForeignKey('changeset_comments.comment_id'))
modified_at = Column('modified_at', DateTime(), nullable=False, default=datetime.datetime.now)
version = Column('version', Integer(), nullable=False, default=0)
comment = relationship('ChangesetComment', lazy='joined')
self.status, self.author
def get_status_lbl(cls, value):
return dict(cls.STATUSES).get(value)
def status_lbl(self):
return ChangesetStatus.get_status_lbl(self.status)
class PullRequest(Base, BaseModel):
__tablename__ = 'pull_requests'
STATUS_NEW = u'new'
STATUS_OPEN = u'open'
STATUS_CLOSED = u'closed'
pull_request_id = Column('pull_request_id', Integer(), nullable=False, primary_key=True)
title = Column('title', Unicode(256), nullable=True)
description = Column('description', UnicodeText(10240), nullable=True)
status = Column('status', Unicode(256), nullable=False, default=STATUS_NEW)
updated_on = Column('updated_on', DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
_revisions = Column('revisions', UnicodeText(20500)) # 500 revisions max
org_repo_id = Column('org_repo_id', Integer(), ForeignKey('repositories.repo_id'), nullable=False)
org_ref = Column('org_ref', Unicode(256), nullable=False)
other_repo_id = Column('other_repo_id', Integer(), ForeignKey('repositories.repo_id'), nullable=False)
other_ref = Column('other_ref', Unicode(256), nullable=False)
def revisions(self):
return self._revisions.split(':')
@revisions.setter
def revisions(self, val):
self._revisions = ':'.join(val)
def org_ref_parts(self):
return self.org_ref.split(':')
def other_ref_parts(self):
return self.other_ref.split(':')
reviewers = relationship('PullRequestReviewers',
org_repo = relationship('Repository', primaryjoin='PullRequest.org_repo_id==Repository.repo_id')
other_repo = relationship('Repository', primaryjoin='PullRequest.other_repo_id==Repository.repo_id')
statuses = relationship('ChangesetStatus')
comments = relationship('ChangesetComment',
def is_closed(self):
return self.status == self.STATUS_CLOSED
def last_review_status(self):
return self.statuses[-1].status if self.statuses else ''
revisions=self.revisions
class PullRequestReviewers(Base, BaseModel):
__tablename__ = 'pull_request_reviewers'
def __init__(self, user=None, pull_request=None):
self.user = user
self.pull_request = pull_request
Status change: