##// END OF EJS Templates
celery: always count tasks even if they fail
celery: always count tasks even if they fail

File last commit:

r4873:d49e3bd8 default
r4891:2cd8759f default
Show More
request.py
109 lines | 3.2 KiB | text/x-python | PythonLexer
# -*- coding: utf-8 -*-
# Copyright (C) 2017-2020 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# 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 Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
from uuid import uuid4
import pyramid.testing
from pyramid.decorator import reify
from pyramid.request import Request as _Request
from rhodecode.translation import _ as tsf
from rhodecode.lib.utils2 import StrictAttributeDict
class TemplateArgs(StrictAttributeDict):
pass
# Base Class with DummyMethods, testing / CLI scripts
class RequestBase(object):
_req_id_bucket = list()
_call_context = TemplateArgs()
_call_context.visual = TemplateArgs()
_call_context.visual.show_sha_length = 12
_call_context.visual.show_revision_number = True
@reify
def req_id(self):
return str(uuid4())
@property
def req_id_bucket(self):
return self._req_id_bucket
def req_id_records_init(self):
self._req_id_bucket = list()
def translate(self, *args, **kwargs):
raise NotImplementedError()
def plularize(self, *args, **kwargs):
raise NotImplementedError()
def get_partial_renderer(self, tmpl_name):
raise NotImplementedError()
@property
def call_context(self):
return self._call_context
def set_call_context(self, new_context):
self._call_context = new_context
# for thin non-web/cli etc
class ThinRequest(RequestBase, pyramid.testing.DummyRequest):
def translate(self, msg):
return msg
def plularize(self, singular, plural, n):
return singular
def get_partial_renderer(self, tmpl_name):
from rhodecode.lib.partial_renderer import get_partial_renderer
return get_partial_renderer(request=self, tmpl_name=tmpl_name)
# for real-web-based
class RealRequest(RequestBase, _Request):
def get_partial_renderer(self, tmpl_name):
from rhodecode.lib.partial_renderer import get_partial_renderer
return get_partial_renderer(request=self, tmpl_name=tmpl_name)
def request_count(self):
from rhodecode.lib.request_counter import get_request_counter
return get_request_counter()
def plularize(self, *args, **kwargs):
return self.localizer.pluralize(*args, **kwargs)
def translate(self, *args, **kwargs):
localizer = self.localizer
def auto_translate(*_args, **_kwargs):
return localizer.translate(tsf(*_args, **_kwargs))
return auto_translate(*args, **kwargs)
class Request(RealRequest):
"""
This is the main request object used in web-context
"""
pass