kallithea Changeset - 6b176c679896

Changeset - 6b176c679896

Parent rev.

Child rev.

[Not reviewed]

beta

0 9 0

Marcin Kuzminski - 13 years ago 2012-09-17 22:17:25
marcin@python-works.com

failsafe the GET `page` argument

9 files changed with 36 insertions and 11 deletions:

rhodecode/controllers/admin/admin.py

rhodecode/controllers/admin/notifications.py

rhodecode/controllers/changelog.py

rhodecode/controllers/followers.py

rhodecode/controllers/forks.py

rhodecode/controllers/journal.py

rhodecode/controllers/search.py

rhodecode/controllers/shortlog.py

rhodecode/lib/utils2.py

0 comments (0 inline, 0 general)

rhodecode/controllers/admin/admin.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.controllers.admin.admin
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     Controller for Admin panel of Rhodecode
     :created_on: Apr 7, 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
 from pylons import request, tmpl_context as c
 from sqlalchemy.orm import joinedload
 from webhelpers.paginate import Page
 from rhodecode.lib.auth import LoginRequired, HasPermissionAllDecorator
 from rhodecode.lib.base import BaseController, render
 from rhodecode.model.db import UserLog
 from rhodecode.lib.utils2 import safe_int
 log = logging.getLogger(__name__)
 class AdminController(BaseController):
     @LoginRequired()
     def __before__(self):
         super(AdminController, self).__before__()
     @HasPermissionAllDecorator('hg.admin')
     def index(self):
         users_log = UserLog.query()\
                 .options(joinedload(UserLog.user))\
                 .options(joinedload(UserLog.repository))\
                 .order_by(UserLog.action_date.desc())
-        p = int(request.params.get('page', 1))
+        p = safe_int(request.params.get('page', 1), 1)
         c.users_log = Page(users_log, page=p, items_per_page=10)
         c.log_data = render('admin/admin_log.html')
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             return c.log_data
         return render('admin/admin.html')

rhodecode/controllers/admin/notifications.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.controllers.admin.notifications
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     notifications controller for RhodeCode
     :created_on: Nov 23, 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 traceback
 from pylons import request
 from pylons import tmpl_context as c, url
 from pylons.controllers.util import redirect
 from webhelpers.paginate import Page
 from rhodecode.lib.base import BaseController, render
 from rhodecode.model.db import Notification
 from rhodecode.model.notification import NotificationModel
 from rhodecode.lib.auth import LoginRequired, NotAnonymous
 from rhodecode.lib import helpers as h
 from rhodecode.model.meta import Session
 from rhodecode.lib.utils2 import safe_int
 log = logging.getLogger(__name__)
 class NotificationsController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     # To properly map this controller, ensure your config/routing.py
     # file has a resource setup:
     #     map.resource('notification', 'notifications', controller='_admin/notifications',
     #         path_prefix='/_admin', name_prefix='_admin_')
     @LoginRequired()
     @NotAnonymous()
     def __before__(self):
         super(NotificationsController, self).__before__()
     def index(self, format='html'):
         """GET /_admin/notifications: All items in the collection"""
         # url('notifications')
         c.user = self.rhodecode_user
         notif = NotificationModel().get_for_user(self.rhodecode_user.user_id,
                                             filter_=request.GET.getall('type'))
         p = int(request.params.get('page', 1))
         p = safe_int(request.params.get('page', 1), 1)
         c.notifications = Page(notif, page=p, items_per_page=10)
         c.pull_request_type = Notification.TYPE_PULL_REQUEST
         c.comment_type = [Notification.TYPE_CHANGESET_COMMENT,
                           Notification.TYPE_PULL_REQUEST_COMMENT]
         _current_filter = request.GET.getall('type')
         c.current_filter = 'all'
         if _current_filter == [c.pull_request_type]:
             c.current_filter = 'pull_request'
         elif _current_filter == c.comment_type:
             c.current_filter = 'comment'
         return render('admin/notifications/notifications.html')
     def mark_all_read(self):
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             nm = NotificationModel()
             # mark all read
             nm.mark_all_read_for_user(self.rhodecode_user.user_id,
                                       filter_=request.GET.getall('type'))
             Session().commit()
             c.user = self.rhodecode_user
             notif = nm.get_for_user(self.rhodecode_user.user_id,
                                     filter_=request.GET.getall('type'))
             c.notifications = Page(notif, page=1, items_per_page=10)
             return render('admin/notifications/notifications_data.html')
     def create(self):
         """POST /_admin/notifications: Create a new item"""
         # url('notifications')
     def new(self, format='html'):
         """GET /_admin/notifications/new: Form to create a new item"""
         # url('new_notification')
     def update(self, notification_id):
         """PUT /_admin/notifications/id: Update an existing item"""
         # Forms posted to this method should contain a hidden field:
         #    <input type="hidden" name="_method" value="PUT" />
         # Or using helpers:
         #    h.form(url('notification', notification_id=ID),
         #           method='put')
         # url('notification', notification_id=ID)
         try:
             no = Notification.get(notification_id)
             owner = lambda: (no.notifications_to_users.user.user_id
                              == c.rhodecode_user.user_id)
             if h.HasPermissionAny('hg.admin')() or owner:
                     NotificationModel().mark_read(c.rhodecode_user.user_id, no)
                     Session().commit()
                     return 'ok'
         except Exception:
             Session.rollback()
             log.error(traceback.format_exc())
         return 'fail'
     def delete(self, notification_id):
         """DELETE /_admin/notifications/id: Delete an existing item"""
         # Forms posted to this method should contain a hidden field:
         #    <input type="hidden" name="_method" value="DELETE" />
         # Or using helpers:
         #    h.form(url('notification', notification_id=ID),
         #           method='delete')
         # url('notification', notification_id=ID)
         try:
             no = Notification.get(notification_id)
             owner = lambda: (no.notifications_to_users.user.user_id
                              == c.rhodecode_user.user_id)
             if h.HasPermissionAny('hg.admin')() or owner:
                     NotificationModel().delete(c.rhodecode_user.user_id, no)
                     Session().commit()
                     return 'ok'
         except Exception:
             Session.rollback()
             log.error(traceback.format_exc())
         return 'fail'
     def show(self, notification_id, format='html'):
         """GET /_admin/notifications/id: Show a specific item"""
         # url('notification', notification_id=ID)
         c.user = self.rhodecode_user
         no = Notification.get(notification_id)
         owner = lambda: (no.notifications_to_users.user.user_id
                          == c.user.user_id)
         if no and (h.HasPermissionAny('hg.admin', 'repository.admin')() or owner):
             unotification = NotificationModel()\
                             .get_user_notification(c.user.user_id, no)
             # if this association to user is not valid, we don't want to show
             # this message
             if unotification:
                 if unotification.read is False:
                     unotification.mark_as_read()
                     Session().commit()
                 c.notification = no
                 return render('admin/notifications/show_notification.html')
         return redirect(url('notifications'))
     def edit(self, notification_id, format='html'):
         """GET /_admin/notifications/id/edit: Form to edit an existing item"""
         # url('edit_notification', notification_id=ID)

rhodecode/controllers/changelog.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.controllers.changelog
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     changelog controller for rhodecode
     :created_on: Apr 21, 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 traceback
 from pylons import request, url, session, tmpl_context as c
 from pylons.controllers.util import redirect
 from pylons.i18n.translation import _
 import rhodecode.lib.helpers as h
 from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator
 from rhodecode.lib.base import BaseRepoController, render
 from rhodecode.lib.helpers import RepoPage
 from rhodecode.lib.compat import json
 from rhodecode.lib.graphmod import _colored, _dagwalker
 from rhodecode.lib.vcs.exceptions import RepositoryError, ChangesetDoesNotExistError
 from rhodecode.lib.utils2 import safe_int
 log = logging.getLogger(__name__)
 class ChangelogController(BaseRepoController):
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def __before__(self):
         super(ChangelogController, self).__before__()
         c.affected_files_cut_off = 60
     def index(self):
         limit = 100
         default = 20
         if request.params.get('size'):
             try:
                 int_size = int(request.params.get('size'))
             except ValueError:
                 int_size = default
             c.size = max(min(int_size, limit), 1)
             session['changelog_size'] = c.size
             session.save()
         else:
             c.size = int(session.get('changelog_size', default))
         # min size must be 1
         c.size = max(c.size, 1)
-        p = int(request.params.get('page', 1))
+        p = safe_int(request.params.get('page', 1), 1)
         branch_name = request.params.get('branch', None)
         try:
             if branch_name:
                 collection = [z for z in
                               c.rhodecode_repo.get_changesets(start=0,
                                                     branch_name=branch_name)]
                 c.total_cs = len(collection)
             else:
                 collection = c.rhodecode_repo
                 c.total_cs = len(c.rhodecode_repo)
             c.pagination = RepoPage(collection, page=p, item_count=c.total_cs,
                                     items_per_page=c.size, branch=branch_name)
             collection = list(c.pagination)
             page_revisions = [x.raw_id for x in collection]
             c.comments = c.rhodecode_db_repo.get_comments(page_revisions)
             c.statuses = c.rhodecode_db_repo.statuses(page_revisions)
         except (RepositoryError, ChangesetDoesNotExistError, Exception), e:
             log.error(traceback.format_exc())
             h.flash(str(e), category='warning')
             return redirect(url('home'))
         self._graph(c.rhodecode_repo, collection, c.total_cs, c.size, p)
         c.branch_name = branch_name
         c.branch_filters = [('', _('All Branches'))] + \
             [(k, k) for k in c.rhodecode_repo.branches.keys()]
         return render('changelog/changelog.html')
     def changelog_details(self, cs):
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             c.cs = c.rhodecode_repo.get_changeset(cs)
             return render('changelog/changelog_details.html')
     def _graph(self, repo, collection, repo_size, size, p):
         """
         Generates a DAG graph for mercurial
         :param repo: repo instance
         :param size: number of commits to show
         :param p: page number
         """
         if not collection:
             c.jsdata = json.dumps([])
             return
         data = []
         revs = [x.revision for x in collection]
         dag = _dagwalker(repo, revs, repo.alias)
         dag = _colored(dag)
         for (id, type, ctx, vtx, edges) in dag:
             data.append(['', vtx, edges])
         c.jsdata = json.dumps(data)

rhodecode/controllers/followers.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.controllers.followers
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     Followers controller for rhodecode
     :created_on: Apr 23, 2011
     :author: marcink
     :copyright: (C) 2011-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
 from pylons import tmpl_context as c, request
 from rhodecode.lib.helpers import Page
 from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator
 from rhodecode.lib.base import BaseRepoController, render
 from rhodecode.model.db import Repository, User, UserFollowing
 from rhodecode.lib.utils2 import safe_int
 log = logging.getLogger(__name__)
 class FollowersController(BaseRepoController):
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def __before__(self):
         super(FollowersController, self).__before__()
     def followers(self, repo_name):
-        p = int(request.params.get('page', 1))
+        p = safe_int(request.params.get('page', 1), 1)
         repo_id = c.rhodecode_db_repo.repo_id
         d = UserFollowing.get_repo_followers(repo_id)\
             .order_by(UserFollowing.follows_from)
         c.followers_pager = Page(d, page=p, items_per_page=20)
         c.followers_data = render('/followers/followers_data.html')
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             return c.followers_data
         return render('/followers/followers.html')

rhodecode/controllers/forks.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.controllers.forks
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     forks controller for rhodecode
     :created_on: Apr 23, 2011
     :author: marcink
     :copyright: (C) 2011-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 traceback
 from formencode import htmlfill
 from pylons import tmpl_context as c, request, url
 from pylons.controllers.util import redirect
 from pylons.i18n.translation import _
 import rhodecode.lib.helpers as h
 from rhodecode.lib.helpers import Page
 from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator, \
     NotAnonymous, HasRepoPermissionAny, HasPermissionAllDecorator,\
     HasPermissionAnyDecorator
 from rhodecode.lib.base import BaseRepoController, render
 from rhodecode.model.db import Repository, RepoGroup, UserFollowing, User
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.forms import RepoForkForm
 from rhodecode.model.scm import ScmModel
 from rhodecode.lib.utils2 import safe_int
 log = logging.getLogger(__name__)
 class ForksController(BaseRepoController):
     @LoginRequired()
     def __before__(self):
         super(ForksController, self).__before__()
     def __load_defaults(self):
         c.repo_groups = RepoGroup.groups_choices(check_perms=True)
         c.repo_groups_choices = map(lambda k: unicode(k[0]), c.repo_groups)
         choices, c.landing_revs = ScmModel().get_repo_landing_revs()
         c.landing_revs_choices = choices
     def __load_data(self, repo_name=None):
         """
         Load defaults settings for edit, and update
         :param repo_name:
         """
         self.__load_defaults()
         c.repo_info = db_repo = Repository.get_by_repo_name(repo_name)
         repo = db_repo.scm_instance
         if c.repo_info is None:
             h.flash(_('%s repository is not mapped to db perhaps'
                       ' it was created or renamed from the filesystem'
                       ' please run the application again'
                       ' in order to rescan repositories') % repo_name,
                       category='error')
             return redirect(url('repos'))
         c.default_user_id = User.get_by_username('default').user_id
         c.in_public_journal = UserFollowing.query()\
             .filter(UserFollowing.user_id == c.default_user_id)\
             .filter(UserFollowing.follows_repository == c.repo_info).scalar()
         if c.repo_info.stats:
             last_rev = c.repo_info.stats.stat_on_revision+1
         else:
             last_rev = 0
         c.stats_revision = last_rev
         c.repo_last_rev = repo.count() if repo.revisions else 0
         if last_rev == 0 or c.repo_last_rev == 0:
             c.stats_percentage = 0
         else:
             c.stats_percentage = '%.2f' % ((float((last_rev)) /
                                             c.repo_last_rev) * 100)
         defaults = RepoModel()._get_defaults(repo_name)
         # add prefix to fork
         defaults['repo_name'] = 'fork-' + defaults['repo_name']
         return defaults
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def forks(self, repo_name):
-        p = int(request.params.get('page', 1))
+        p = safe_int(request.params.get('page', 1), 1)
         repo_id = c.rhodecode_db_repo.repo_id
         d = []
         for r in Repository.get_repo_forks(repo_id):
             if not HasRepoPermissionAny(
                 'repository.read', 'repository.write', 'repository.admin'
             )(r.repo_name, 'get forks check'):
                 continue
             d.append(r)
         c.forks_pager = Page(d, page=p, items_per_page=20)
         c.forks_data = render('/forks/forks_data.html')
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             return c.forks_data
         return render('/forks/forks.html')
     @NotAnonymous()
     @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def fork(self, repo_name):
         c.repo_info = Repository.get_by_repo_name(repo_name)
         if not c.repo_info:
             h.flash(_('%s repository is not mapped to db perhaps'
                       ' it was created or renamed from the file system'
                       ' please run the application again'
                       ' in order to rescan repositories') % repo_name,
                       category='error')
             return redirect(url('home'))
         defaults = self.__load_data(repo_name)
         return htmlfill.render(
             render('forks/fork.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )
     @NotAnonymous()
     @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def fork_create(self, repo_name):
         self.__load_defaults()
         c.repo_info = Repository.get_by_repo_name(repo_name)
         _form = RepoForkForm(old_data={'repo_type': c.repo_info.repo_type},
                              repo_groups=c.repo_groups_choices,
                              landing_revs=c.landing_revs_choices)()
         form_result = {}
         try:
             form_result = _form.to_python(dict(request.POST))
             # create fork is done sometimes async on celery, db transaction
             # management is handled there.
             RepoModel().create_fork(form_result, self.rhodecode_user.user_id)
             h.flash(_('forked %s repository as %s') \
                       % (repo_name, form_result['repo_name']),
                     category='success')
         except formencode.Invalid, errors:
             c.new_repo = errors.value['repo_name']
             return htmlfill.render(
                 render('forks/fork.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8")
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during repository forking %s') %
                     repo_name, category='error')
         return redirect(url('home'))

rhodecode/controllers/journal.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.controllers.journal
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     Journal controller for pylons
     :created_on: Nov 21, 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
 from itertools import groupby
 from sqlalchemy import or_
 from sqlalchemy.orm import joinedload
 from webhelpers.paginate import Page
 from webhelpers.feedgenerator import Atom1Feed, Rss201rev2Feed
 from webob.exc import HTTPBadRequest
 from pylons import request, tmpl_context as c, response, url
 from pylons.i18n.translation import _
 import rhodecode.lib.helpers as h
 from rhodecode.lib.auth import LoginRequired, NotAnonymous
 from rhodecode.lib.base import BaseController, render
 from rhodecode.model.db import UserLog, UserFollowing, Repository, User
 from rhodecode.model.meta import Session
 from sqlalchemy.sql.expression import func
 from rhodecode.model.scm import ScmModel
 from rhodecode.lib.utils2 import safe_int
 log = logging.getLogger(__name__)
 class JournalController(BaseController):
     def __before__(self):
         super(JournalController, self).__before__()
         self.language = 'en-us'
         self.ttl = "5"
         self.feed_nr = 20
     @LoginRequired()
     @NotAnonymous()
     def index(self):
         # Return a rendered template
-        p = int(request.params.get('page', 1))
+        p = safe_int(request.params.get('page', 1), 1)
         c.user = User.get(self.rhodecode_user.user_id)
         all_repos = self.sa.query(Repository)\
                      .filter(Repository.user_id == c.user.user_id)\
                      .order_by(func.lower(Repository.repo_name)).all()
         c.user_repos = ScmModel().get_repos(all_repos)
         c.following = self.sa.query(UserFollowing)\
             .filter(UserFollowing.user_id == self.rhodecode_user.user_id)\
             .options(joinedload(UserFollowing.follows_repository))\
             .all()
         journal = self._get_journal_data(c.following)
         c.journal_pager = Page(journal, page=p, items_per_page=20)
         c.journal_day_aggreagate = self._get_daily_aggregate(c.journal_pager)
         c.journal_data = render('journal/journal_data.html')
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             return c.journal_data
         return render('journal/journal.html')
     @LoginRequired(api_access=True)
     @NotAnonymous()
     def journal_atom(self):
         """
         Produce an atom-1.0 feed via feedgenerator module
         """
         following = self.sa.query(UserFollowing)\
             .filter(UserFollowing.user_id == self.rhodecode_user.user_id)\
             .options(joinedload(UserFollowing.follows_repository))\
             .all()
         return self._atom_feed(following, public=False)
     @LoginRequired(api_access=True)
     @NotAnonymous()
     def journal_rss(self):
         """
         Produce an rss feed via feedgenerator module
         """
         following = self.sa.query(UserFollowing)\
             .filter(UserFollowing.user_id == self.rhodecode_user.user_id)\
             .options(joinedload(UserFollowing.follows_repository))\
             .all()
         return self._rss_feed(following, public=False)
     def _get_daily_aggregate(self, journal):
         groups = []
         for k, g in groupby(journal, lambda x: x.action_as_day):
             user_group = []
             for k2, g2 in groupby(list(g), lambda x: x.user.email):
                 l = list(g2)
                 user_group.append((l[0].user, l))
             groups.append((k, user_group,))
         return groups
     def _get_journal_data(self, following_repos):
         repo_ids = [x.follows_repository.repo_id for x in following_repos
                     if x.follows_repository is not None]
         user_ids = [x.follows_user.user_id for x in following_repos
                     if x.follows_user is not None]
         filtering_criterion = None
         if repo_ids and user_ids:
             filtering_criterion = or_(UserLog.repository_id.in_(repo_ids),
                         UserLog.user_id.in_(user_ids))
         if repo_ids and not user_ids:
             filtering_criterion = UserLog.repository_id.in_(repo_ids)
         if not repo_ids and user_ids:
             filtering_criterion = UserLog.user_id.in_(user_ids)
         if filtering_criterion is not None:
             journal = self.sa.query(UserLog)\
                 .options(joinedload(UserLog.user))\
                 .options(joinedload(UserLog.repository))\
                 .filter(filtering_criterion)\
                 .order_by(UserLog.action_date.desc())
         else:
             journal = []
         return journal
     @LoginRequired()
     @NotAnonymous()
     def toggle_following(self):
         cur_token = request.POST.get('auth_token')
         token = h.get_token()
         if cur_token == token:
             user_id = request.POST.get('follows_user_id')
             if user_id:
                 try:
                     self.scm_model.toggle_following_user(user_id,
                                                 self.rhodecode_user.user_id)
                     Session.commit()
                     return 'ok'
                 except:
                     raise HTTPBadRequest()
             repo_id = request.POST.get('follows_repo_id')
             if repo_id:
                 try:
                     self.scm_model.toggle_following_repo(repo_id,
                                                 self.rhodecode_user.user_id)
                     Session.commit()
                     return 'ok'
                 except:
                     raise HTTPBadRequest()
         log.debug('token mismatch %s vs %s' % (cur_token, token))
         raise HTTPBadRequest()
     @LoginRequired()
     def public_journal(self):
         # Return a rendered template
-        p = int(request.params.get('page', 1))
+        p = safe_int(request.params.get('page', 1), 1)
         c.following = self.sa.query(UserFollowing)\
             .filter(UserFollowing.user_id == self.rhodecode_user.user_id)\
             .options(joinedload(UserFollowing.follows_repository))\
             .all()
         journal = self._get_journal_data(c.following)
         c.journal_pager = Page(journal, page=p, items_per_page=20)
         c.journal_day_aggreagate = self._get_daily_aggregate(c.journal_pager)
         c.journal_data = render('journal/journal_data.html')
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             return c.journal_data
         return render('journal/public_journal.html')
     def _atom_feed(self, repos, public=True):
         journal = self._get_journal_data(repos)
         if public:
             _link = url('public_journal_atom', qualified=True)
             _desc = '%s %s %s' % (c.rhodecode_name, _('public journal'),
                                   'atom feed')
         else:
             _link = url('journal_atom', qualified=True)
             _desc = '%s %s %s' % (c.rhodecode_name, _('journal'), 'atom feed')
         feed = Atom1Feed(title=_desc,
                          link=_link,
                          description=_desc,
                          language=self.language,
                          ttl=self.ttl)
         for entry in journal[:self.feed_nr]:
             action, action_extra, ico = h.action_parser(entry, feed=True)
             title = "%s - %s %s" % (entry.user.short_contact, action(),
                                  entry.repository.repo_name)
             desc = action_extra()
             _url = None
             if entry.repository is not None:
                 _url = url('changelog_home',
                            repo_name=entry.repository.repo_name,
                            qualified=True)
             feed.add_item(title=title,
                           pubdate=entry.action_date,
                           link=_url or url('', qualified=True),
                           author_email=entry.user.email,
                           author_name=entry.user.full_contact,
                           description=desc)
         response.content_type = feed.mime_type
         return feed.writeString('utf-8')
     def _rss_feed(self, repos, public=True):
         journal = self._get_journal_data(repos)
         if public:
             _link = url('public_journal_atom', qualified=True)
             _desc = '%s %s %s' % (c.rhodecode_name, _('public journal'),
                                   'rss feed')
         else:
             _link = url('journal_atom', qualified=True)
             _desc = '%s %s %s' % (c.rhodecode_name, _('journal'), 'rss feed')
         feed = Rss201rev2Feed(title=_desc,
                          link=_link,
                          description=_desc,
                          language=self.language,
                          ttl=self.ttl)
         for entry in journal[:self.feed_nr]:
             action, action_extra, ico = h.action_parser(entry, feed=True)
             title = "%s - %s %s" % (entry.user.short_contact, action(),
                                  entry.repository.repo_name)
             desc = action_extra()
             _url = None
             if entry.repository is not None:
                 _url = url('changelog_home',
                            repo_name=entry.repository.repo_name,
                            qualified=True)
             feed.add_item(title=title,
                           pubdate=entry.action_date,
                           link=_url or url('', qualified=True),
                           author_email=entry.user.email,
                           author_name=entry.user.full_contact,
                           description=desc)
         response.content_type = feed.mime_type
         return feed.writeString('utf-8')
     @LoginRequired(api_access=True)
     def public_journal_atom(self):
         """
         Produce an atom-1.0 feed via feedgenerator module
         """
         c.following = self.sa.query(UserFollowing)\
             .filter(UserFollowing.user_id == self.rhodecode_user.user_id)\
             .options(joinedload(UserFollowing.follows_repository))\
             .all()
         return self._atom_feed(c.following)
     @LoginRequired(api_access=True)
     def public_journal_rss(self):
         """
         Produce an rss2 feed via feedgenerator module
         """
         c.following = self.sa.query(UserFollowing)\
             .filter(UserFollowing.user_id == self.rhodecode_user.user_id)\
             .options(joinedload(UserFollowing.follows_repository))\
             .all()
         return self._rss_feed(c.following)

rhodecode/controllers/search.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.controllers.search
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     Search controller for RhodeCode
     :created_on: Aug 7, 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 traceback
 from pylons.i18n.translation import _
 from pylons import request, config, tmpl_context as c
 from rhodecode.lib.auth import LoginRequired
 from rhodecode.lib.base import BaseController, render
 from rhodecode.lib.indexers import CHGSETS_SCHEMA, SCHEMA, CHGSET_IDX_NAME, \
     IDX_NAME, WhooshResultWrapper
 from webhelpers.paginate import Page
 from webhelpers.util import update_params
 from whoosh.index import open_dir, EmptyIndexError
 from whoosh.qparser import QueryParser, QueryParserError
 from whoosh.query import Phrase, Wildcard, Term, Prefix
 from rhodecode.model.repo import RepoModel
 from rhodecode.lib.utils2 import safe_str
+from rhodecode.lib.utils2 import safe_str, safe_int
 log = logging.getLogger(__name__)
 class SearchController(BaseController):
     @LoginRequired()
     def __before__(self):
         super(SearchController, self).__before__()
     def index(self, search_repo=None):
         c.repo_name = search_repo
         c.formated_results = []
         c.runtime = ''
         c.cur_query = request.GET.get('q', None)
         c.cur_type = request.GET.get('type', 'content')
         c.cur_search = search_type = {'content': 'content',
                                       'commit': 'message',
                                       'path': 'path',
                                       'repository': 'repository'
                                       }.get(c.cur_type, 'content')
         index_name = {
             'content': IDX_NAME,
             'commit': CHGSET_IDX_NAME,
             'path': IDX_NAME
         }.get(c.cur_type, IDX_NAME)
         schema_defn = {
             'content': SCHEMA,
             'commit': CHGSETS_SCHEMA,
             'path': SCHEMA
         }.get(c.cur_type, SCHEMA)
         log.debug('IDX: %s' % index_name)
         log.debug('SCHEMA: %s' % schema_defn)
         if c.cur_query:
             cur_query = c.cur_query.lower()
             log.debug(cur_query)
         if c.cur_query:
-            p = int(request.params.get('page', 1))
+            p = safe_int(request.params.get('page', 1), 1)
             highlight_items = set()
             try:
                 idx = open_dir(config['app_conf']['index_dir'],
                                indexname=index_name)
                 searcher = idx.searcher()
                 qp = QueryParser(search_type, schema=schema_defn)
                 if c.repo_name:
                     cur_query = u'repository:%s %s' % (c.repo_name, cur_query)
                 try:
                     query = qp.parse(unicode(cur_query))
                     # extract words for highlight
                     if isinstance(query, Phrase):
                         highlight_items.update(query.words)
                     elif isinstance(query, Prefix):
                         highlight_items.add(query.text)
                     else:
                         for i in query.all_terms():
                             if i[0] in ['content', 'message']:
                                 highlight_items.add(i[1])
                     matcher = query.matcher(searcher)
                     log.debug('query: %s' % query)
                     log.debug('hl terms: %s' % highlight_items)
                     results = searcher.search(query)
                     res_ln = len(results)
                     c.runtime = '%s results (%.3f seconds)' % (
                         res_ln, results.runtime
+                    )
                     def url_generator(**kw):
                         return update_params("?q=%s&type=%s" \
                         % (safe_str(c.cur_query), safe_str(c.cur_type)), **kw)
                     repo_location = RepoModel().repos_path
                     c.formated_results = Page(
                         WhooshResultWrapper(search_type, searcher, matcher,
                                             highlight_items, repo_location),
                         page=p,
                         item_count=res_ln,
                         items_per_page=10,
                         url=url_generator
+                    )
                 except QueryParserError:
                     c.runtime = _('Invalid search query. Try quoting it.')
                 searcher.close()
             except (EmptyIndexError, IOError):
                 log.error(traceback.format_exc())
                 log.error('Empty Index data')
                 c.runtime = _('There is no index to search in. '
                               'Please run whoosh indexer')
             except (Exception):
                 log.error(traceback.format_exc())
                 c.runtime = _('An error occurred during this search operation')
         # Return a rendered template
         return render('/search/search.html')

rhodecode/controllers/shortlog.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.controllers.shortlog
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     Shortlog controller for rhodecode
     :created_on: Apr 18, 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
 from pylons import tmpl_context as c, request, url
 from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator
 from rhodecode.lib.base import BaseRepoController, render
 from rhodecode.lib.helpers import RepoPage
 from pylons.controllers.util import redirect
 from rhodecode.lib.utils2 import safe_int
 log = logging.getLogger(__name__)
 class ShortlogController(BaseRepoController):
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def __before__(self):
         super(ShortlogController, self).__before__()
     def index(self, repo_name):
         p = int(request.params.get('page', 1))
         size = int(request.params.get('size', 20))
         p = safe_int(request.params.get('page', 1), 1)
         size = safe_int(request.params.get('size', 20), 20)
         def url_generator(**kw):
             return url('shortlog_home', repo_name=repo_name, size=size, **kw)
         c.repo_changesets = RepoPage(c.rhodecode_repo, page=p,
                                     items_per_page=size, url=url_generator)
         if not c.repo_changesets:
             return redirect(url('summary_home', repo_name=repo_name))
         c.shortlog_data = render('shortlog/shortlog_data.html')
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             return c.shortlog_data
         r = render('shortlog/shortlog.html')
         return r

rhodecode/lib/utils2.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.lib.utils
     ~~~~~~~~~~~~~~~~~~~
     Some simple helper functions
     :created_on: Jan 5, 2011
     :author: marcink
     :copyright: (C) 2011-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 re
 import time
 import datetime
 from pylons.i18n.translation import _, ungettext
 from rhodecode.lib.vcs.utils.lazy import LazyProperty
 def __get_lem():
     """
     Get language extension map based on what's inside pygments lexers
     """
     from pygments import lexers
     from string import lower
     from collections import defaultdict
     d = defaultdict(lambda: [])
     def __clean(s):
         s = s.lstrip('*')
         s = s.lstrip('.')
         if s.find('[') != -1:
             exts = []
             start, stop = s.find('['), s.find(']')
             for suffix in s[start + 1:stop]:
                 exts.append(s[:s.find('[')] + suffix)
             return map(lower, exts)
         else:
             return map(lower, [s])
     for lx, t in sorted(lexers.LEXERS.items()):
         m = map(__clean, t[-2])
         if m:
             m = reduce(lambda x, y: x + y, m)
             for ext in m:
                 desc = lx.replace('Lexer', '')
                 d[ext].append(desc)
     return dict(d)
 def str2bool(_str):
     """
     returs True/False value from given string, it tries to translate the
     string into boolean
     :param _str: string value to translate into boolean
     :rtype: boolean
     :returns: boolean from given string
     """
     if _str is None:
         return False
     if _str in (True, False):
         return _str
     _str = str(_str).strip().lower()
     return _str in ('t', 'true', 'y', 'yes', 'on', '1')
 def convert_line_endings(line, mode):
     """
     Converts a given line  "line end" accordingly to given mode
     Available modes are::
 - Unix
 - Mac
 - DOS
     :param line: given line to convert
     :param mode: mode to convert to
     :rtype: str
     :return: converted line according to mode
     """
     from string import replace
     if mode == 0:
             line = replace(line, '\r\n', '\n')
             line = replace(line, '\r', '\n')
     elif mode == 1:
             line = replace(line, '\r\n', '\r')
             line = replace(line, '\n', '\r')
     elif mode == 2:
             line = re.sub("\r(?!\n)|(?<!\r)\n", "\r\n", line)
     return line
 def detect_mode(line, default):
     """
     Detects line break for given line, if line break couldn't be found
     given default value is returned
     :param line: str line
     :param default: default
     :rtype: int
     :return: value of line end on of 0 - Unix, 1 - Mac, 2 - DOS
     """
     if line.endswith('\r\n'):
         return 2
     elif line.endswith('\n'):
         return 0
     elif line.endswith('\r'):
         return 1
     else:
         return default
 def generate_api_key(username, salt=None):
     """
     Generates unique API key for given username, if salt is not given
     it'll be generated from some random string
     :param username: username as string
     :param salt: salt to hash generate KEY
     :rtype: str
     :returns: sha1 hash from username+salt
     """
     from tempfile import _RandomNameSequence
     import hashlib
     if salt is None:
         salt = _RandomNameSequence().next()
     return hashlib.sha1(username + salt).hexdigest()
 def safe_int(val, default=None):
     """
     Returns int() of val if val is not convertable to int use default
     instead
     :param val:
     :param default:
     """
     try:
         val = int(val)
     except ValueError:
         val = default
     return val
 def safe_unicode(str_, from_encoding=None):
     """
     safe unicode function. Does few trick to turn str_ into unicode
     In case of UnicodeDecode error we try to return it with encoding detected
     by chardet library if it fails fallback to unicode with errors replaced
     :param str_: string to decode
     :rtype: unicode
     :returns: unicode object
     """
     if isinstance(str_, unicode):
         return str_
     if not from_encoding:
         import rhodecode
         DEFAULT_ENCODING = rhodecode.CONFIG.get('default_encoding','utf8')
         from_encoding = DEFAULT_ENCODING
     try:
         return unicode(str_)
     except UnicodeDecodeError:
         pass
     try:
         return unicode(str_, from_encoding)
     except UnicodeDecodeError:
         pass
     try:
         import chardet
         encoding = chardet.detect(str_)['encoding']
         if encoding is None:
             raise Exception()
         return str_.decode(encoding)
     except (ImportError, UnicodeDecodeError, Exception):
         return unicode(str_, from_encoding, 'replace')
 def safe_str(unicode_, to_encoding=None):
     """
     safe str function. Does few trick to turn unicode_ into string
     In case of UnicodeEncodeError we try to return it with encoding detected
     by chardet library if it fails fallback to string with errors replaced
     :param unicode_: unicode to encode
     :rtype: str
     :returns: str object
     """
     # if it's not basestr cast to str
     if not isinstance(unicode_, basestring):
         return str(unicode_)
     if isinstance(unicode_, str):
         return unicode_
     if not to_encoding:
         import rhodecode
         DEFAULT_ENCODING = rhodecode.CONFIG.get('default_encoding','utf8')
         to_encoding = DEFAULT_ENCODING
     try:
         return unicode_.encode(to_encoding)
     except UnicodeEncodeError:
         pass
     try:
         import chardet
         encoding = chardet.detect(unicode_)['encoding']
         if encoding is None:
             raise UnicodeEncodeError()
         return unicode_.encode(encoding)
     except (ImportError, UnicodeEncodeError):
         return unicode_.encode(to_encoding, 'replace')
     return safe_str
 def engine_from_config(configuration, prefix='sqlalchemy.', **kwargs):
     """
     Custom engine_from_config functions that makes sure we use NullPool for
     file based sqlite databases. This prevents errors on sqlite. This only
     applies to sqlalchemy versions < 0.7.0
     """
     import sqlalchemy
     from sqlalchemy import engine_from_config as efc
     import logging
     if int(sqlalchemy.__version__.split('.')[1]) < 7:
         # This solution should work for sqlalchemy < 0.7.0, and should use
         # proxy=TimerProxy() for execution time profiling
         from sqlalchemy.pool import NullPool
         url = configuration[prefix + 'url']
         if url.startswith('sqlite'):
             kwargs.update({'poolclass': NullPool})
         return efc(configuration, prefix, **kwargs)
     else:
         import time
         from sqlalchemy import event
         from sqlalchemy.engine import Engine
         log = logging.getLogger('sqlalchemy.engine')
         BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE = xrange(30, 38)
         engine = efc(configuration, prefix, **kwargs)
         def color_sql(sql):
             COLOR_SEQ = "\033[1;%dm"
             COLOR_SQL = YELLOW
             normal = '\x1b[0m'
             return ''.join([COLOR_SEQ % COLOR_SQL, sql, normal])
         if configuration['debug']:
             #attach events only for debug configuration
             def before_cursor_execute(conn, cursor, statement,
                                     parameters, context, executemany):
                 context._query_start_time = time.time()
                 log.info(color_sql(">>>>> STARTING QUERY >>>>>"))
             def after_cursor_execute(conn, cursor, statement,
                                     parameters, context, executemany):
                 total = time.time() - context._query_start_time
                 log.info(color_sql("<<<<< TOTAL TIME: %f <<<<<" % total))
             event.listen(engine, "before_cursor_execute",
                          before_cursor_execute)
             event.listen(engine, "after_cursor_execute",
                          after_cursor_execute)
     return engine
 def age(prevdate):
     """
     turns a datetime into an age string.
     :param prevdate: datetime object
     :rtype: unicode
     :returns: unicode words describing age
     """
     order = ['year', 'month', 'day', 'hour', 'minute', 'second']
     deltas = {}
     # Get date parts deltas
     now = datetime.datetime.now()
     for part in order:
         deltas[part] = getattr(now, part) - getattr(prevdate, part)
     # Fix negative offsets (there is 1 second between 10:59:59 and 11:00:00,
     # not 1 hour, -59 minutes and -59 seconds)
     for num, length in [(5, 60), (4, 60), (3, 24)]:  # seconds, minutes, hours
         part = order[num]
         carry_part = order[num - 1]
         if deltas[part] < 0:
             deltas[part] += length
             deltas[carry_part] -= 1
     # Same thing for days except that the increment depends on the (variable)
     # number of days in the month
     month_lengths = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
     if deltas['day'] < 0:
         if prevdate.month == 2 and (prevdate.year % 4 == 0 and
             (prevdate.year % 100 != 0 or prevdate.year % 400 == 0)):
             deltas['day'] += 29
         else:
             deltas['day'] += month_lengths[prevdate.month - 1]
         deltas['month'] -= 1
     if deltas['month'] < 0:
         deltas['month'] += 12
         deltas['year'] -= 1
     # Format the result
     fmt_funcs = {
         'year': lambda d: ungettext(u'%d year', '%d years', d) % d,
         'month': lambda d: ungettext(u'%d month', '%d months', d) % d,
         'day': lambda d: ungettext(u'%d day', '%d days', d) % d,
         'hour': lambda d: ungettext(u'%d hour', '%d hours', d) % d,
         'minute': lambda d: ungettext(u'%d minute', '%d minutes', d) % d,
         'second': lambda d: ungettext(u'%d second', '%d seconds', d) % d,
+    }
     for i, part in enumerate(order):
         value = deltas[part]
         if value == 0:
             continue
         if i < 5:
             sub_part = order[i + 1]
             sub_value = deltas[sub_part]
         else:
             sub_value = 0
         if sub_value == 0:
             return _(u'%s ago') % fmt_funcs[part](value)
         return _(u'%s and %s ago') % (fmt_funcs[part](value),
             fmt_funcs[sub_part](sub_value))
     return _(u'just now')
 def uri_filter(uri):
     """
     Removes user:password from given url string
     :param uri:
     :rtype: unicode
     :returns: filtered list of strings
     """
     if not uri:
         return ''
     proto = ''
     for pat in ('https://', 'http://'):
         if uri.startswith(pat):
             uri = uri[len(pat):]
             proto = pat
             break
     # remove passwords and username
     uri = uri[uri.find('@') + 1:]
     # get the port
     cred_pos = uri.find(':')
     if cred_pos == -1:
         host, port = uri, None
     else:
         host, port = uri[:cred_pos], uri[cred_pos + 1:]
     return filter(None, [proto, host, port])
 def credentials_filter(uri):
     """
     Returns a url with removed credentials
     :param uri:
     """
     uri = uri_filter(uri)
     #check if we have port
     if len(uri) > 2 and uri[2]:
         uri[2] = ':' + uri[2]
     return ''.join(uri)
 def get_changeset_safe(repo, rev):
     """
     Safe version of get_changeset if this changeset doesn't exists for a
     repo it returns a Dummy one instead
     :param repo:
     :param rev:
     """
     from rhodecode.lib.vcs.backends.base import BaseRepository
     from rhodecode.lib.vcs.exceptions import RepositoryError
     from rhodecode.lib.vcs.backends.base import EmptyChangeset
     if not isinstance(repo, BaseRepository):
         raise Exception('You must pass an Repository '
                         'object as first argument got %s', type(repo))
     try:
         cs = repo.get_changeset(rev)
     except RepositoryError:
         cs = EmptyChangeset(requested_revision=rev)
     return cs
 def datetime_to_time(dt):
     if dt:
         return time.mktime(dt.timetuple())
 def time_to_datetime(tm):
     if tm:
         if isinstance(tm, basestring):
             try:
                 tm = float(tm)
             except ValueError:
                 return
         return datetime.datetime.fromtimestamp(tm)
 MENTIONS_REGEX = r'(?:^@|\s@)([a-zA-Z0-9]{1}[a-zA-Z0-9\-\_\.]+)(?:\s{1})'
 def extract_mentioned_users(s):
     """
     Returns unique usernames from given string s that have @mention
     :param s: string to get mentions
     """
     usrs = set()
     for username in re.findall(MENTIONS_REGEX, s):
         usrs.add(username)
     return sorted(list(usrs), key=lambda k: k.lower())
 class AttributeDict(dict):
     def __getattr__(self, attr):
         return self.get(attr, None)
     __setattr__ = dict.__setitem__
     __delattr__ = dict.__delitem__

0 comments (0 inline, 0 general)