celery: move async tasks to model - they both use model and are used by model, so they must be pretty much the same

kallithea/controllers/admin/settings.py

@@ -39,12 +39,11 b' import kallithea'
 from kallithea.lib import webutils
 from kallithea.lib.auth import HasPermissionAnyDecorator, LoginRequired
 from kallithea.lib.base import BaseController, render
-from kallithea.lib.celerylib import tasks
 from kallithea.lib.utils import repo2db_mapper, set_app_settings
 from kallithea.lib.utils2 import safe_str
 from kallithea.lib.vcs import VCSError
 from kallithea.lib.webutils import url
-from kallithea.model import db, meta
+from kallithea.model import async_tasks, db, meta
 from kallithea.model.forms import ApplicationSettingsForm, ApplicationUiSettingsForm, ApplicationVisualisationForm
 from kallithea.model.notification import EmailNotificationModel
 from kallithea.model.scm import ScmModel
@@ -301,7 +300,7 b' class SettingsController(BaseController)'
 
             recipients = [test_email] if test_email else None
 
-            tasks.send_email(recipients, test_email_subj,
+            async_tasks.send_email(recipients, test_email_subj,
                              test_email_txt_body, test_email_html_body)
 
             webutils.flash(_('Send email task created'), category='success')
@@ -379,7 +378,7 b' class SettingsController(BaseController)'
         if request.POST:
             repo_location = self._get_hg_ui_settings()['paths_root_path']
             full_index = request.POST.get('full_index', False)
-            tasks.whoosh_index(repo_location, full_index)
+            async_tasks.whoosh_index(repo_location, full_index)
             webutils.flash(_('Whoosh reindex task scheduled'), category='success')
             raise HTTPFound(location=url('admin_settings_search'))
 

kallithea/controllers/summary.py

@@ -41,7 +41,6 b' from webob.exc import HTTPBadRequest'
 from kallithea.lib import ext_json, webutils
 from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseRepoController, jsonify, render
-from kallithea.lib.celerylib.tasks import get_commits_stats
 from kallithea.lib.conf import ALL_EXTS, ALL_READMES, LANGUAGES_EXTENSIONS_MAP
 from kallithea.lib.markup_renderer import MarkupRenderer
 from kallithea.lib.page import Page
@@ -49,7 +48,7 b' from kallithea.lib.utils2 import safe_in'
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.lib.vcs.exceptions import ChangesetError, EmptyRepositoryError, NodeDoesNotExistError
 from kallithea.lib.vcs.nodes import FileNode
-from kallithea.model import db
+from kallithea.model import async_tasks, db
 
 
 log = logging.getLogger(__name__)
@@ -209,5 +208,5 b' class SummaryController(BaseRepoControll'
             c.trending_languages = []
 
         recurse_limit = 500  # don't recurse more than 500 times when parsing
-        get_commits_stats(c.db_repo.repo_name, ts_min_y, ts_max_y, recurse_limit)
+        async_tasks.get_commits_stats(c.db_repo.repo_name, ts_min_y, ts_max_y, recurse_limit)
         return render('summary/statistics.html')

kallithea/lib/celerypylons/__init__.py

@@ -23,7 +23,7 b' import kallithea'
 
 
 class CeleryConfig(object):
-    imports = ['kallithea.lib.celerylib.tasks']
+    imports = ['kallithea.model.async_tasks']
     task_always_eager = False
 
 # map from Kallithea .ini Celery 3 config names to Celery 4 config names

kallithea/model/async_tasks.py

@@ -12,8 +12,8 b''
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
-kallithea.lib.celerylib.tasks
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+kallithea.model.async_tasks
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 Kallithea task modules, containing all task that suppose to be run
 by celery daemon
@@ -45,8 +45,7 b' from kallithea.lib import celerylib, con'
 from kallithea.lib.indexers.daemon import WhooshIndexingDaemon
 from kallithea.lib.utils2 import asbool, ascii_bytes
 from kallithea.lib.vcs.utils import author_email
-from kallithea.model import db, userlog
-from kallithea.model.repo import RepoModel
+from kallithea.model import db, repo, userlog
 
 
 __all__ = ['whoosh_index', 'get_commits_stats', 'send_email']
@@ -87,12 +86,12 b' def get_commits_stats(repo_name, ts_min_'
 
         co_day_auth_aggr = {}
         commits_by_day_aggregate = {}
-        repo = db.Repository.get_by_repo_name(repo_name)
-        if repo is None:
+        db_repo = db.Repository.get_by_repo_name(repo_name)
+        if db_repo is None:
             return True
 
-        repo = repo.scm_instance
-        repo_size = repo.count()
+        scm_repo = db_repo.scm_instance
+        repo_size = scm_repo.count()
         # return if repo have no revisions
         if repo_size < 1:
             lock.release()
@@ -112,7 +111,7 b' def get_commits_stats(repo_name, ts_min_'
         if cur_stats is not None:
             last_rev = cur_stats.stat_on_revision
 
-        if last_rev == repo.get_changeset().revision and repo_size > 1:
+        if last_rev == scm_repo.get_changeset().revision and repo_size > 1:
             # pass silently without any work if we're not on first revision or
             # current state of parsing revision(from db marker) is the
             # last revision
@@ -130,7 +129,7 b' def get_commits_stats(repo_name, ts_min_'
         log.debug('Getting revisions from %s to %s',
              last_rev, last_rev + parse_limit
         )
-        for cs in repo[last_rev:last_rev + parse_limit]:
+        for cs in scm_repo[last_rev:last_rev + parse_limit]:
             log.debug('parsing %s', cs)
             last_cs = cs  # remember last parsed changeset
             tt = cs.date.timetuple()
@@ -188,8 +187,8 b' def get_commits_stats(repo_name, ts_min_'
                                key=itemgetter(0))
 
         if not co_day_auth_aggr:
-            co_day_auth_aggr[akc(repo.contact)] = {
-                "label": akc(repo.contact),
+            co_day_auth_aggr[akc(scm_repo.contact)] = {
+                "label": akc(scm_repo.contact),
                 "data": [0, 1],
                 "schema": ["commits"],
             }
@@ -199,7 +198,7 b' def get_commits_stats(repo_name, ts_min_'
         stats.commit_activity_combined = ascii_bytes(ext_json.dumps(overview_data))
 
         log.debug('last revision %s', last_rev)
-        leftovers = len(repo.revisions[last_rev:])
+        leftovers = len(scm_repo.revisions[last_rev:])
         log.debug('revisions to parse %s', leftovers)
 
         if last_rev == 0 or leftovers < parse_limit:
@@ -221,7 +220,7 b' def get_commits_stats(repo_name, ts_min_'
         lock.release()
 
         # execute another task if celery is enabled
-        if len(repo.revisions) > 1 and kallithea.CELERY_APP and recurse_limit > 0:
+        if len(scm_repo.revisions) > 1 and kallithea.CELERY_APP and recurse_limit > 0:
             get_commits_stats(repo_name, ts_min_y, ts_max_y, recurse_limit - 1)
         elif recurse_limit <= 0:
             log.debug('Not recursing - limit has been reached')
@@ -376,7 +375,7 b' def create_repo(form_data, cur_user):'
     enable_downloads = defs.get('repo_enable_downloads')
 
     try:
-        repo = RepoModel()._create_repo(
+        db_repo = repo.RepoModel()._create_repo(
             repo_name=repo_name_full,
             repo_type=repo_type,
             description=description,
@@ -398,30 +397,30 b' def create_repo(form_data, cur_user):'
 
         DBS.commit()
         # now create this repo on Filesystem
-        RepoModel()._create_filesystem_repo(
+        repo.RepoModel()._create_filesystem_repo(
             repo_name=repo_name,
             repo_type=repo_type,
             repo_group=db.RepoGroup.guess_instance(repo_group),
             clone_uri=clone_uri,
         )
-        repo = db.Repository.get_by_repo_name(repo_name_full)
-        hooks.log_create_repository(repo.get_dict(), created_by=owner.username)
+        db_repo = db.Repository.get_by_repo_name(repo_name_full)
+        hooks.log_create_repository(db_repo.get_dict(), created_by=owner.username)
 
         # update repo changeset caches initially
-        repo.update_changeset_cache()
+        db_repo.update_changeset_cache()
 
         # set new created state
-        repo.set_state(db.Repository.STATE_CREATED)
+        db_repo.set_state(db.Repository.STATE_CREATED)
         DBS.commit()
     except Exception as e:
         log.warning('Exception %s occurred when forking repository, '
                     'doing cleanup...' % e)
         # rollback things manually !
-        repo = db.Repository.get_by_repo_name(repo_name_full)
-        if repo:
-            db.Repository.delete(repo.repo_id)
+        db_repo = db.Repository.get_by_repo_name(repo_name_full)
+        if db_repo:
+            db.Repository.delete(db_repo.repo_id)
             DBS.commit()
-            RepoModel()._delete_filesystem_repo(repo)
+            repo.RepoModel()._delete_filesystem_repo(db_repo)
         raise
 
     return True
@@ -455,7 +454,7 b' def create_repo_fork(form_data, cur_user'
     try:
         fork_of = db.Repository.guess_instance(form_data.get('fork_parent_id'))
 
-        RepoModel()._create_repo(
+        repo.RepoModel()._create_repo(
             repo_name=repo_name_full,
             repo_type=repo_type,
             description=form_data['description'],
@@ -474,39 +473,39 b' def create_repo_fork(form_data, cur_user'
         source_repo_path = os.path.join(base_path, fork_of.repo_name)
 
         # now create this repo on Filesystem
-        RepoModel()._create_filesystem_repo(
+        repo.RepoModel()._create_filesystem_repo(
             repo_name=repo_name,
             repo_type=repo_type,
             repo_group=db.RepoGroup.guess_instance(repo_group),
             clone_uri=source_repo_path,
         )
-        repo = db.Repository.get_by_repo_name(repo_name_full)
-        hooks.log_create_repository(repo.get_dict(), created_by=owner.username)
+        db_repo = db.Repository.get_by_repo_name(repo_name_full)
+        hooks.log_create_repository(db_repo.get_dict(), created_by=owner.username)
 
         # update repo changeset caches initially
-        repo.update_changeset_cache()
+        db_repo.update_changeset_cache()
 
         # set new created state
-        repo.set_state(db.Repository.STATE_CREATED)
+        db_repo.set_state(db.Repository.STATE_CREATED)
         DBS.commit()
     except Exception as e:
         log.warning('Exception %s occurred when forking repository, '
                     'doing cleanup...' % e)
         # rollback things manually !
-        repo = db.Repository.get_by_repo_name(repo_name_full)
-        if repo:
-            db.Repository.delete(repo.repo_id)
+        db_repo = db.Repository.get_by_repo_name(repo_name_full)
+        if db_repo:
+            db.Repository.delete(db_repo.repo_id)
             DBS.commit()
-            RepoModel()._delete_filesystem_repo(repo)
+            repo.RepoModel()._delete_filesystem_repo(db_repo)
         raise
 
     return True
 
 
 def __get_codes_stats(repo_name):
-    repo = db.Repository.get_by_repo_name(repo_name).scm_instance
+    scm_repo = db.Repository.get_by_repo_name(repo_name).scm_instance
 
-    tip = repo.get_changeset()
+    tip = scm_repo.get_changeset()
     code_stats = {}
 
     for _topnode, _dirnodes, filenodes in tip.walk('/'):

kallithea/model/notification.py

@@ -34,7 +34,7 b' from tg import tmpl_context as c'
 from tg.i18n import ugettext as _
 
 from kallithea.lib.utils2 import fmt_date
-from kallithea.model import db
+from kallithea.model import async_tasks, db
 
 
 log = logging.getLogger(__name__)
@@ -66,7 +66,6 b' class NotificationModel(object):'
         :param email_kwargs: additional dict to pass as args to email template
         """
         import kallithea.lib.helpers as h
-        from kallithea.lib.celerylib import tasks
         email_kwargs = email_kwargs or {}
         if recipients and not getattr(recipients, '__iter__', False):
             raise Exception('recipients must be a list or iterable')
@@ -135,7 +134,7 b' class NotificationModel(object):'
 
         # send email with notification to participants
         for rec_mail in sorted(rec_mails):
-            tasks.send_email([rec_mail], email_subject, email_txt_body,
+            async_tasks.send_email([rec_mail], email_subject, email_txt_body,
                      email_html_body, headers,
                      from_name=created_by_obj.full_name_or_username)
 

kallithea/model/repo.py

@@ -405,8 +405,8 b' class RepoModel(object):'
         :param form_data:
         :param cur_user:
         """
-        from kallithea.lib.celerylib import tasks
-        return tasks.create_repo(form_data, cur_user)
+        from kallithea.model import async_tasks
+        return async_tasks.create_repo(form_data, cur_user)
 
     def _update_permissions(self, repo, perms_new=None, perms_updates=None,
                             check_perms=True):
@@ -448,8 +448,8 b' class RepoModel(object):'
         :param form_data:
         :param cur_user:
         """
-        from kallithea.lib.celerylib import tasks
-        return tasks.create_repo_fork(form_data, cur_user)
+        from kallithea.model import async_tasks
+        return async_tasks.create_repo_fork(form_data, cur_user)
 
     def delete(self, repo, forks=None, fs_remove=True, cur_user=None):
         """

kallithea/model/user.py

@@ -298,8 +298,7 b' class UserModel(object):'
         allowing users to copy-paste or manually enter the token from the
         email.
         """
-        from kallithea.lib.celerylib import tasks
-        from kallithea.model import notification
+        from kallithea.model import async_tasks, notification
 
         user_email = data['email']
         user = db.User.get_by_email(user_email)
@@ -332,7 +331,7 b' class UserModel(object):'
                 reset_token=token,
                 reset_url=link)
             log.debug('sending email')
-            tasks.send_email([user_email], _("Password reset link"), body, html_body)
+            async_tasks.send_email([user_email], _("Password reset link"), body, html_body)
             log.info('send new password mail to %s', user_email)
         else:
             log.debug("password reset email %s not found", user_email)
@@ -365,7 +364,7 b' class UserModel(object):'
         return expected_token == token
 
     def reset_password(self, user_email, new_passwd):
-        from kallithea.lib.celerylib import tasks
+        from kallithea.model import async_tasks
         user = db.User.get_by_email(user_email)
         if user is not None:
             if not self.can_change_password(user):
@@ -376,7 +375,7 b' class UserModel(object):'
         if new_passwd is None:
             raise Exception('unable to set new password')
 
-        tasks.send_email([user_email],
+        async_tasks.send_email([user_email],
                  _('Password reset notification'),
                  _('The password to your account %s has been changed using password reset form.') % (user.username,))
         log.info('send password reset mail to %s', user_email)

kallithea/tests/functional/test_login.py

@@ -6,7 +6,7 b' import urllib.parse'
 import mock
 from tg.util.webtest import test_context
 
-import kallithea.lib.celerylib.tasks
+import kallithea.model.async_tasks
 from kallithea.lib import webutils
 from kallithea.lib.utils2 import check_password, generate_api_key
 from kallithea.model import db, meta, validators
@@ -410,7 +410,7 b' class TestLoginController(base.TestContr'
         def mock_send_email(recipients, subject, body='', html_body='', headers=None, from_name=None):
             collected.append((recipients, subject, body, html_body))
 
-        with mock.patch.object(kallithea.lib.celerylib.tasks, 'send_email', mock_send_email), \
+        with mock.patch.object(kallithea.model.async_tasks, 'send_email', mock_send_email), \
                 mock.patch.object(time, 'time', lambda: timestamp):
             response = self.app.post(base.url(controller='login',
                                          action='password_reset'),

kallithea/tests/models/test_notifications.py

@@ -5,8 +5,8 b' import mock'
 from tg.util.webtest import test_context
 
 import kallithea.lib.celerylib
-import kallithea.lib.celerylib.tasks
 import kallithea.lib.helpers as h
+import kallithea.model.async_tasks
 from kallithea.model import db, meta
 from kallithea.model.notification import EmailNotificationModel, NotificationModel
 from kallithea.model.user import UserModel
@@ -48,7 +48,7 b' class TestNotifications(base.TestControl'
                 assert body == "hi there"
                 assert '>hi there<' in html_body
                 assert from_name == 'u1 u1'
-            with mock.patch.object(kallithea.lib.celerylib.tasks, 'send_email', send_email):
+            with mock.patch.object(kallithea.model.async_tasks, 'send_email', send_email):
                 NotificationModel().create(created_by=self.u1,
                                                    body='hi there',
                                                    recipients=usrs)
@@ -73,7 +73,7 b' class TestNotifications(base.TestControl'
             l.append('<hr/>\n')
 
         with test_context(self.app):
-            with mock.patch.object(kallithea.lib.celerylib.tasks, 'send_email', send_email):
+            with mock.patch.object(kallithea.model.async_tasks, 'send_email', send_email):
                 pr_kwargs = dict(
                     pr_nice_id='#7',
                     pr_title='The Title',
@@ -155,7 +155,7 b' class TestNotifications(base.TestControl'
                 # Email type TYPE_PASSWORD_RESET has no corresponding notification type - test it directly:
                 desc = 'TYPE_PASSWORD_RESET'
                 kwargs = dict(user='John Doe', reset_token='decbf64715098db5b0bd23eab44bd792670ab746', reset_url='http://reset.com/decbf64715098db5b0bd23eab44bd792670ab746')
-                kallithea.lib.celerylib.tasks.send_email(['[e-mail]'],
+                kallithea.model.async_tasks.send_email(['[e-mail]'],
                     "Password reset link",
                     EmailNotificationModel().get_email_tmpl(EmailNotificationModel.TYPE_PASSWORD_RESET, 'txt', **kwargs),
                     EmailNotificationModel().get_email_tmpl(EmailNotificationModel.TYPE_PASSWORD_RESET, 'html', **kwargs),

kallithea/tests/other/test_mail.py

@@ -25,7 +25,7 b' class smtplib_mock(object):'
         smtplib_mock.lastmsg = msg
 
 
[e-mail]('kallithea.lib.celerylib.tasks.smtplib', smtplib_mock)
[e-mail]('kallithea.model.async_tasks.smtplib', smtplib_mock)
 class TestMail(base.TestController):
 
     def test_send_mail_trivial(self):
@@ -40,8 +40,8 b' class TestMail(base.TestController):'
             'smtp_server': mailserver,
             'app_email_from': envelope_from,
         }
-        with mock.patch('kallithea.lib.celerylib.tasks.config', config_mock):
-            kallithea.lib.celerylib.tasks.send_email(recipients, subject, body, html_body)
+        with mock.patch('kallithea.model.async_tasks.config', config_mock):
+            kallithea.model.async_tasks.send_email(recipients, subject, body, html_body)
 
         assert smtplib_mock.lastdest == set(recipients)
         assert smtplib_mock.lastsender == envelope_from
@@ -64,8 +64,8 b' class TestMail(base.TestController):'
             'app_email_from': envelope_from,
             'email_to': email_to,
         }
-        with mock.patch('kallithea.lib.celerylib.tasks.config', config_mock):
-            kallithea.lib.celerylib.tasks.send_email(recipients, subject, body, html_body)
+        with mock.patch('kallithea.model.async_tasks.config', config_mock):
+            kallithea.model.async_tasks.send_email(recipients, subject, body, html_body)
 
         assert smtplib_mock.lastdest == set([base.TEST_USER_ADMIN_EMAIL, email_to])
         assert smtplib_mock.lastsender == envelope_from
@@ -88,8 +88,8 b' class TestMail(base.TestController):'
             'app_email_from': envelope_from,
             'email_to': email_to,
         }
-        with mock.patch('kallithea.lib.celerylib.tasks.config', config_mock):
-            kallithea.lib.celerylib.tasks.send_email(recipients, subject, body, html_body)
+        with mock.patch('kallithea.model.async_tasks.config', config_mock):
+            kallithea.model.async_tasks.send_email(recipients, subject, body, html_body)
 
         assert smtplib_mock.lastdest == set([base.TEST_USER_ADMIN_EMAIL] + email_to.split(','))
         assert smtplib_mock.lastsender == envelope_from
@@ -110,8 +110,8 b' class TestMail(base.TestController):'
             'smtp_server': mailserver,
             'app_email_from': envelope_from,
         }
-        with mock.patch('kallithea.lib.celerylib.tasks.config', config_mock):
-            kallithea.lib.celerylib.tasks.send_email(recipients, subject, body, html_body)
+        with mock.patch('kallithea.model.async_tasks.config', config_mock):
+            kallithea.model.async_tasks.send_email(recipients, subject, body, html_body)
 
         assert smtplib_mock.lastdest == set([base.TEST_USER_ADMIN_EMAIL])
         assert smtplib_mock.lastsender == envelope_from
@@ -133,8 +133,8 b' class TestMail(base.TestController):'
             'smtp_server': mailserver,
             'app_email_from': envelope_from,
         }
-        with mock.patch('kallithea.lib.celerylib.tasks.config', config_mock):
-            kallithea.lib.celerylib.tasks.send_email(recipients, subject, body, html_body, from_name=author.full_name_or_username)
+        with mock.patch('kallithea.model.async_tasks.config', config_mock):
+            kallithea.model.async_tasks.send_email(recipients, subject, body, html_body, from_name=author.full_name_or_username)
 
         assert smtplib_mock.lastdest == set(recipients)
         assert smtplib_mock.lastsender == envelope_from
@@ -157,8 +157,8 b' class TestMail(base.TestController):'
             'smtp_server': mailserver,
             'app_email_from': envelope_from,
         }
-        with mock.patch('kallithea.lib.celerylib.tasks.config', config_mock):
-            kallithea.lib.celerylib.tasks.send_email(recipients, subject, body, html_body, from_name=author.full_name_or_username)
+        with mock.patch('kallithea.model.async_tasks.config', config_mock):
+            kallithea.model.async_tasks.send_email(recipients, subject, body, html_body, from_name=author.full_name_or_username)
 
         assert smtplib_mock.lastdest == set(recipients)
         assert smtplib_mock.lastsender == envelope_from
@@ -181,8 +181,8 b' class TestMail(base.TestController):'
             'smtp_server': mailserver,
             'app_email_from': envelope_from,
         }
-        with mock.patch('kallithea.lib.celerylib.tasks.config', config_mock):
-            kallithea.lib.celerylib.tasks.send_email(recipients, subject, body, html_body,
+        with mock.patch('kallithea.model.async_tasks.config', config_mock):
+            kallithea.model.async_tasks.send_email(recipients, subject, body, html_body,
                                                      from_name=author.full_name_or_username, headers=headers)
 
         assert smtplib_mock.lastdest == set(recipients)

scripts/deps.py

@@ -131,7 +131,6 b' conftest'
 
 normal_modules = set('''
 kallithea
-kallithea.lib.celerylib.tasks
 kallithea.lib
 kallithea.lib.auth
 kallithea.lib.auth_modules
@@ -146,6 +145,7 b' kallithea.lib.utils2'
 kallithea.lib.vcs
 kallithea.lib.webutils
 kallithea.model
+kallithea.model.async_tasks
 kallithea.model.scm
 kallithea.templates.py
 '''.split())
@@ -158,10 +158,10 b' known_violations = ['
 ('kallithea.lib.utils', 'kallithea.model'),  # clean up utils
 ('kallithea.lib.utils', 'kallithea.model.db'),
 ('kallithea.lib.utils', 'kallithea.model.scm'),
-('kallithea.lib.celerylib.tasks', 'kallithea.lib.helpers'),
-('kallithea.lib.celerylib.tasks', 'kallithea.lib.hooks'),
-('kallithea.lib.celerylib.tasks', 'kallithea.lib.indexers'),
-('kallithea.lib.celerylib.tasks', 'kallithea.model'),
+('kallithea.model.async_tasks', 'kallithea.lib.helpers'),
+('kallithea.model.async_tasks', 'kallithea.lib.hooks'),
+('kallithea.model.async_tasks', 'kallithea.lib.indexers'),
+('kallithea.model.async_tasks', 'kallithea.model'),
 ('kallithea.model', 'kallithea.lib.auth'),  # auth.HasXXX
 ('kallithea.model', 'kallithea.lib.auth_modules'),  # validators
 ('kallithea.model', 'kallithea.lib.helpers'),