unrhodecode/rhodecode/lib/celerylib/tasks.py

# Copyright (C) 2012-2024 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/

"""
RhodeCode task modules, containing all task that suppose to be run
by celery daemon
"""

import os
import time
from copy import deepcopy
from logging import Logger
from typing import Any

from celery import current_app
from pyramid_mailer.mailer import Mailer
from pyramid_mailer.mailer import DebugMailer
from pyramid_mailer.message import Message
from email.utils import formatdate

import rhodecode
from rhodecode.apps.ai_agents.ai_service import get_ai_service
from rhodecode.apps.ai_agents.models.base import Response, AIServiceError
from rhodecode.lib import audit_logger, diffs, codeblocks
from rhodecode.lib.celerylib import get_logger, async_task, RequestContextTask, run_task
from rhodecode.lib import hooks_base
from rhodecode.lib.diffs import MAX_CONTEXT
from rhodecode.lib.str_utils import safe_str
from rhodecode.lib.utils import adopt_for_celery
from rhodecode.lib.utils2 import safe_int, str2bool, aslist
from rhodecode.lib.statsd_client import StatsdClient
from rhodecode.model.comment import CommentsModel
from rhodecode.model.db import (
    true,
    null,
    Session,
    IntegrityError,
    Repository,
    RepoGroup,
    User,
    ScheduleEntry,
    PullRequest,
)
from rhodecode.model.permission import PermissionModel
from rhodecode.model.pull_request import PullRequestModel
from rhodecode.model.settings import SettingsModel


@async_task(ignore_result=True, base=RequestContextTask)
def send_email(recipients, subject, body="", html_body="", email_config=None, extra_headers=None):
    """
    Sends an email with defined parameters from the .ini files.

    :param recipients: list of recipients, it this is empty the defined email
        address from field 'email_to' is used instead
    :param subject: subject of the mail
    :param body: body of the mail
    :param html_body: html version of body
    :param email_config: specify custom configuration for mailer
    :param extra_headers: specify custom headers
    """
    log = get_logger(send_email)

    email_config = email_config or rhodecode.CONFIG

    mail_server = email_config.get("smtp_server") or None
    dev_mod = str2bool(email_config.get("development_email"))

    if mail_server is None and not dev_mod:
        log.error(
            "SMTP server information missing. Sending email failed. "
            "Make sure that `smtp_server` variable is configured "
            "inside the .ini file"
        )
        return False
    conf_prefix = email_config.get("email_prefix", None)
    prefix = f"{conf_prefix} " if conf_prefix else ""
    subject = f"{prefix}{subject}"

    if recipients:
        if isinstance(recipients, str):
            recipients = recipients.split(",")
    else:
        # if recipients are not defined we send to email_config + all admins
        admins = []
        for u in User.query().filter(User.admin == true()).all():
            if u.email:
                admins.append(u.email)
        recipients = []
        config_email = email_config.get("email_to")
        if config_email:
            recipients += [config_email]
        recipients += admins

    # translate our LEGACY config into the one that pyramid_mailer supports
    transformed_email_confing = transform_legacy_email_config(email_config, mail_server)

    if extra_headers is None:
        extra_headers = {}

    extra_headers.setdefault("Date", formatdate(time.time()))

    if "thread_ids" in extra_headers:
        thread_ids = extra_headers.pop("thread_ids")
        extra_headers["References"] = " ".join(f"<{t}>" for t in thread_ids)

    try:
        mailer = get_mailer(transformed_email_confing, email_config)

        message = Message(
            subject=subject,
            sender=transformed_email_confing["default_sender"],
            recipients=recipients,
            body=body,
            html=html_body,
            extra_headers=extra_headers,
        )
        mailer.send_immediately(message)

        statsd = StatsdClient.statsd
        if statsd:
            statsd.incr("rhodecode_email_sent_total")

    except Exception:
        log.exception("Mail sending failed")
        return False
    return True


def transform_legacy_email_config(email_config: dict[Any, Any] | Any, mail_server: Any | None) -> dict[
    str, None | int | bool | str | Any]:
    return dict(
        host=mail_server,
        port=email_config.get("smtp_port", 25),
        username=email_config.get("smtp_username", None),
        password=email_config.get("smtp_password", None),
        tls=str2bool(email_config.get("smtp_use_tls")),
        ssl=str2bool(email_config.get("smtp_use_ssl")),
        # SSL key file
        # keyfile='',
        # SSL certificate file
        # certfile='',
        # Location of maildir
        # queue_path='',
        default_sender=email_config.get("app_email_from", "RhodeCode-noreply@rhodecode.com"),
        debug=str2bool(email_config.get("smtp_debug")),
        # /usr/sbin/sendmail	Sendmail executable
        # sendmail_app='',
        # {sendmail_app} -t -i -f {sender}	Template for sendmail execution
        # sendmail_template='',
    )


def get_mailer(transformed_email_conf: dict[str, Any], original_email_conf: dict[str, Any]) -> Mailer | DebugMailer:
    log = get_logger(get_mailer)
    dev_mode = str2bool(original_email_conf.get("development_email"))

    if dev_mode:
        emails_path = original_email_conf.get("local_email_store") or "./local_emails"
        log.debug(f"Getting development email. emails folder path: {emails_path}")
        return DebugMailer(emails_path)
    return Mailer(**transformed_email_conf)


@async_task(ignore_result=True, base=RequestContextTask)
def create_repo(form_data, cur_user):
    from rhodecode.model.repo import RepoModel
    from rhodecode.model.user import UserModel
    from rhodecode.model.scm import ScmModel
    from rhodecode.model.settings import SettingsModel

    log = get_logger(create_repo)

    cur_user = UserModel()._get_user(cur_user)
    owner = cur_user

    repo_name = form_data["repo_name"]
    repo_name_full = form_data["repo_name_full"]
    repo_type = form_data["repo_type"]
    description = form_data["repo_description"]
    private = form_data["repo_private"]
    clone_uri = form_data.get("clone_uri")
    repo_group = safe_int(form_data["repo_group"])
    copy_fork_permissions = form_data.get("copy_permissions")
    copy_group_permissions = form_data.get("repo_copy_permissions")
    fork_of = form_data.get("fork_parent_id")
    state = form_data.get("repo_state", Repository.STATE_PENDING)

    # repo creation defaults, private and repo_type are filled in form
    defs = SettingsModel().get_default_repo_settings(strip_prefix=True)
    enable_statistics = form_data.get("enable_statistics", defs.get("repo_enable_statistics"))
    enable_locking = form_data.get("enable_locking", defs.get("repo_enable_locking"))
    enable_downloads = form_data.get("enable_downloads", defs.get("repo_enable_downloads"))

    # set landing rev based on default branches for SCM
    if repo_type == "git":
        # Get default branch from global Git settings
        from rhodecode.model.settings import VcsSettingsModel

        settings_model = VcsSettingsModel()
        global_settings = settings_model.get_global_settings()
        default_branch = global_settings.get("vcs_git_default_branch", "master")
        landing_ref = f"branch:{default_branch}"
    else:
        landing_ref, _label = ScmModel.backend_landing_ref(repo_type)

    try:
        RepoModel()._create_repo(
            repo_name=repo_name_full,
            repo_type=repo_type,
            description=description,
            owner=owner,
            private=private,
            clone_uri=clone_uri,
            repo_group=repo_group,
            landing_rev=landing_ref,
            fork_of=fork_of,
            copy_fork_permissions=copy_fork_permissions,
            copy_group_permissions=copy_group_permissions,
            enable_statistics=enable_statistics,
            enable_locking=enable_locking,
            enable_downloads=enable_downloads,
            state=state,
        )

        Session().commit()

        # now create this repo on Filesystem
        RepoModel()._create_filesystem_repo(
            repo_name=repo_name,
            repo_type=repo_type,
            repo_group=RepoModel()._get_repo_group(repo_group),
            clone_uri=clone_uri,
        )
        repo = Repository.get_by_repo_name(repo_name_full)
        hooks_base.create_repository(created_by=owner.username, **repo.get_dict())

        # update repo commit caches initially
        repo.update_commit_cache(recursive=False)

        # set new created state
        repo.set_state(Repository.STATE_CREATED)
        repo_id = repo.repo_id
        repo_data = repo.get_api_data()

        audit_logger.store(
            "repo.create",
            action_data={"data": repo_data},
            user=cur_user,
            repo=audit_logger.RepoWrap(repo_name=repo_name, repo_id=repo_id),
        )

        Session().commit()

        PermissionModel().trigger_permission_flush()

    except Exception as e:
        log.warning("Exception occurred when creating repository, doing cleanup...", exc_info=True)
        if isinstance(e, IntegrityError):
            Session().rollback()

        # rollback things manually !
        repo = Repository.get_by_repo_name(repo_name_full)
        if repo:
            Repository.delete(repo.repo_id)
            Session().commit()
            RepoModel()._delete_filesystem_repo(repo)
        log.info("Cleanup of repo %s finished", repo_name_full)
        raise

    return True


@async_task(ignore_result=True, base=RequestContextTask)
def create_repo_fork(form_data, cur_user):
    """
    Creates a fork of repository using internal VCS methods
    """
    from rhodecode.model.repo import RepoModel
    from rhodecode.model.user import UserModel

    log = get_logger(create_repo_fork)

    cur_user = UserModel()._get_user(cur_user)
    owner = cur_user

    repo_name = form_data["repo_name"]  # fork in this case
    repo_name_full = form_data["repo_name_full"]
    repo_type = form_data["repo_type"]
    description = form_data["description"]
    private = form_data["private"]
    clone_uri = form_data.get("clone_uri")
    repo_group = safe_int(form_data["repo_group"])
    landing_ref = form_data["landing_rev"]
    copy_fork_permissions = form_data.get("copy_permissions")
    fork_id = safe_int(form_data.get("fork_parent_id"))

    try:
        fork_of = RepoModel()._get_repo(fork_id)
        RepoModel()._create_repo(
            repo_name=repo_name_full,
            repo_type=repo_type,
            description=description,
            owner=owner,
            private=private,
            clone_uri=clone_uri,
            repo_group=repo_group,
            landing_rev=landing_ref,
            fork_of=fork_of,
            copy_fork_permissions=copy_fork_permissions,
        )

        Session().commit()

        base_path = Repository.base_path()
        source_repo_path = os.path.join(base_path, fork_of.repo_name)

        # now create this repo on Filesystem
        RepoModel()._create_filesystem_repo(
            repo_name=repo_name,
            repo_type=repo_type,
            repo_group=RepoModel()._get_repo_group(repo_group),
            clone_uri=source_repo_path,
        )
        repo = Repository.get_by_repo_name(repo_name_full)
        hooks_base.create_repository(created_by=owner.username, **repo.get_dict())

        # update repo commit caches initially
        config = repo._config
        config.set("extensions", "largefiles", "")
        repo.update_commit_cache(config=config, recursive=False)

        # set new created state
        repo.set_state(Repository.STATE_CREATED)

        repo_id = repo.repo_id
        repo_data = repo.get_api_data()
        audit_logger.store(
            "repo.fork",
            action_data={"data": repo_data},
            user=cur_user,
            repo=audit_logger.RepoWrap(repo_name=repo_name, repo_id=repo_id),
        )

        Session().commit()
    except Exception as e:
        log.warning("Exception occurred when forking repository, doing cleanup...", exc_info=True)
        if isinstance(e, IntegrityError):
            Session().rollback()

        # rollback things manually !
        repo = Repository.get_by_repo_name(repo_name_full)
        if repo:
            Repository.delete(repo.repo_id)
            Session().commit()
            RepoModel()._delete_filesystem_repo(repo)
        log.info("Cleanup of repo %s finished", repo_name_full)
        raise

    return True


@async_task(ignore_result=True, base=RequestContextTask)
def repo_maintenance(repoid):
    from rhodecode.lib import repo_maintenance as repo_maintenance_lib

    log = get_logger(repo_maintenance)
    repo = Repository.get_by_id_or_repo_name(repoid)
    if repo:
        maintenance = repo_maintenance_lib.RepoMaintenance()
        tasks = maintenance.get_tasks_for_repo(repo)
        log.debug("Executing %s tasks on repo `%s`", tasks, repoid)
        executed_types = maintenance.execute(repo)
        log.debug("Got execution results %s", executed_types)
    else:
        log.debug("Repo `%s` not found or without a clone_url", repoid)


@async_task(ignore_result=True, base=RequestContextTask)
def check_for_update(send_email_notification=True, email_recipients=None):
    from rhodecode.model.update import UpdateModel
    from rhodecode.model.notification import EmailNotificationModel

    log = get_logger(check_for_update)
    update_url = UpdateModel().get_update_url()
    cur_ver = rhodecode.__version__

    try:
        data = UpdateModel().get_update_data(update_url)

        current_ver = UpdateModel().get_stored_version(fallback=cur_ver)
        latest_ver = data["versions"][0]["version"]
        UpdateModel().store_version(latest_ver)

        if send_email_notification:
            log.debug(
                "Send email notification is enabled. Current RhodeCode version: %s, latest known: %s",
                current_ver,
                latest_ver,
            )
            if UpdateModel().is_outdated(current_ver, latest_ver):
                email_kwargs = {
                    "current_ver": current_ver,
                    "latest_ver": latest_ver,
                }

                (subject, email_body, email_body_plaintext) = EmailNotificationModel().render_email(
                    EmailNotificationModel.TYPE_UPDATE_AVAILABLE, **email_kwargs
                )

                email_recipients = aslist(email_recipients, sep=",") or [
                    user.email for user in User.get_all_super_admins()
                ]
                run_task(send_email, email_recipients, subject, email_body_plaintext, email_body)

    except Exception:
        log.exception("Failed to check for update")
        raise


def sync_last_update_for_objects(*args, **kwargs):
    skip_repos = kwargs.get("skip_repos")
    if not skip_repos:
        repos = Repository.query().order_by(Repository.group_id.asc())

        for repo in repos:
            repo.update_commit_cache(recursive=False)

    skip_groups = kwargs.get("skip_groups")
    if not skip_groups:
        repo_groups = RepoGroup.query().filter(RepoGroup.group_parent_id == null())

        for root_gr in repo_groups:
            for repo_gr in reversed(root_gr.recursive_groups()):
                repo_gr.update_commit_cache()


@async_task(ignore_result=True, base=RequestContextTask)
def test_celery_exception(msg):
    raise Exception(f"Test exception: {msg}")


@async_task(ignore_result=True, base=RequestContextTask)
def sync_last_update(*args, **kwargs):
    sync_last_update_for_objects(*args, **kwargs)


@async_task(ignore_result=False)
def beat_check(*args, **kwargs):
    log = get_logger(beat_check)
    log.info("%r: Got args: %r and kwargs %r", beat_check, args, kwargs)
    return time.time()


@async_task(ignore_result=True)
def schedule_sync_ldap_ad_users_producer():
    log = get_logger(schedule_sync_ldap_ad_users_producer)

    try:
        from rc_ee.lib.celerylib.scheduler import RcScheduler
    except ImportError:
        log.error("Attempt to schedule EE feature")
        return

    scheduler = RcScheduler(app=current_app)
    scheduler.sync()
    task_name = "rc_ee.lib.celerylib.tasks.sync_ldap_ad_users_producer"
    if task_name not in scheduler.schedule:
        entries = {
            task_name: {
                "task": task_name,
                "schedule_type": "crontab",
                "schedule_value": {"hour": 0, "minute": 4},
                "options": {"expires": 12 * 3600},
            }
        }
        scheduler.update_from_dict(entries)


@async_task(ignore_result=True)
def unschedule_sync_ldap_ad_users_producer():
    log = get_logger(unschedule_sync_ldap_ad_users_producer)

    try:
        from rc_ee.lib.celerylib.scheduler import RcScheduler
    except ImportError:
        log.error("Attempt to unschedule EE feature")
        return

    task_name = "rc_ee.lib.celerylib.tasks.sync_ldap_ad_users_producer"
    existing_task = ScheduleEntry.query().filter(ScheduleEntry.task_dot_notation == task_name).first()
    if existing_task:
        Session().delete(existing_task)


@async_task
def start_ai_code_review(pull_request_id):
    log = get_logger(start_ai_code_review)
    log.info("Starting AI code review for pull request id: %s", pull_request_id)
    rc_settings = SettingsModel()
    ai_setting = rc_settings.get_setting_by_name("ai_features_enabled")
    ai_enabled = False
    ai_user = User.get_ai_user()

    if ai_setting:
        ai_enabled = ai_setting.app_settings_value

    if not ai_enabled:
        err_msg = f"AI code review is disabled. Pull request id: {pull_request_id}"
        log.warning(err_msg)
        audit_logger.store(
            "ai.code-review.error",
            user=ai_user,
            action_data={
                audit_logger.PR_ID: pull_request_id,
                audit_logger.AI_MODEL: None,
                "error": True,
                "error_message": err_msg,
            },
        )
        return

    pull_request: PullRequest = PullRequest.get(pull_request_id)

    if not pull_request:
        err_msg = f"Pull request id: {pull_request_id} not found, review not possible."
        log.error(err_msg)
        _set_ai_pr_state(pull_request, "error")
        audit_logger.store(
            "ai.code-review.error",
            user=ai_user,
            action_data={
                audit_logger.PR_ID: pull_request_id,
                audit_logger.AI_MODEL: None,
                "error": True,
                "error_message": err_msg,
            },
        )
        return

    settings_valid, err_msg = _validate_settings(rc_settings)
    if not settings_valid:
        log.error(err_msg)
        _set_ai_pr_state(pull_request, "error")
        audit_logger.store(
            "ai.code-review.error",
            user=ai_user,
            action_data={
                audit_logger.PR_ID: pull_request_id,
                audit_logger.AI_MODEL: None,
                "error": True,
                "error_message": err_msg,
            },
            repo=pull_request.target_repo,
        )
        return

    ai_model = rc_settings.get_setting_by_name("ai_model").app_settings_value
    ai_model_version = rc_settings.get_setting_by_name("ai_model_version").app_settings_value
    ai_api_key = rc_settings.get_setting_by_name("ai_api_key").app_settings_value

    if not ai_model or not ai_model or not ai_api_key:
        log.info("AI model or model version or API key is not set, review not possible.")
        _set_ai_pr_state(pull_request, "error")
        return

    audit_logger.store(
        "ai.code-review.start",
        user=ai_user,
        action_data={
            audit_logger.PR_ID: pull_request_id,
            audit_logger.AI_MODEL: ai_model,
            audit_logger.AI_MODEL_VERSION: ai_model_version,
        },
        repo=pull_request.target_repo,
        commit=True,
    )

    if not pull_request.revisions:
        err_msg = f"Current revision not found for pull request id: {pull_request_id} not found, review not possible."
        log.error(err_msg)
        _set_ai_pr_state(pull_request, "error")
        audit_logger.store(
            "ai.code-review.finish",
            user=ai_user,
            action_data={
                audit_logger.PR_ID: pull_request_id,
                audit_logger.AI_MODEL: ai_model_version,
                "error": True,
                "error_message": err_msg,
            },
            repo=pull_request.target_repo,
        )
        return

    if pull_request.is_state_changing():
        err_msg = f"Pull request id: {pull_request_id} is in state changing, review not possible."
        log.error(err_msg)
        _set_ai_pr_state(pull_request, "error")
        audit_logger.store(
            "ai.code-review.finish",
            user=ai_user,
            action_data={
                audit_logger.PR_ID: pull_request_id,
                audit_logger.AI_MODEL: ai_model_version,
                "error": True,
                "error_message": err_msg,
            },
            repo=pull_request.target_repo,
        )
        return

    diffset = _get_diffset(pull_request)

    try:
        service = get_ai_service(
            api_key=ai_api_key,
            model_name=ai_model,
            api_model_version_name=ai_model_version,
        )

        instructions = rc_settings.get_setting_by_name("ai_code_review_instructions")
        if instructions:
            instructions = instructions.app_settings_value.split("\r\n")
        else:
            log.warning("instructions not set, using default instructions.")
            instructions = service.DEFAULT_BASIC_REVIEW_POINTS

        response = service.code_review(diffset, instructions=instructions)
        _add_comments(response, pull_request, log, ai_user)

        _set_ai_pr_state(pull_request, "finished")

        audit_logger.store(
            "ai.code-review.finish",
            user=ai_user,
            action_data={
                audit_logger.PR_ID: pull_request_id,
                audit_logger.AI_MODEL: response.model,
                "error": response.error,
                "error_message": "",
            },
            repo=pull_request.target_repo,
        )
    except AIServiceError as e:
        log.error("AI service error: %s", e)
        _set_ai_pr_state(pull_request, "error")
        audit_logger.store(
            "ai.code-review.finish",
            user=ai_user,
            action_data={
                audit_logger.PR_ID: pull_request_id,
                audit_logger.AI_MODEL: service.get_api_model_name(),
                "error": True,
                "error_message": str(e),
            },
            repo=pull_request.target_repo,
        )


def _set_ai_pr_state(pull_request: PullRequest, state: str | None):
    ai_state = deepcopy(pull_request.ai_code_review_state)

    if not ai_state:
        ai_state = {
            "review_state": state
        }
    else:
        ai_state["review_state"] = state

    pull_request.ai_code_review_state = ai_state
    Session().commit()


def _add_comments(response: Response, pull_request: PullRequest, log: Logger | Any, ai_user: User):
    log.debug(
        "Response from AI service: %s",
        response.message,
    )
    for response_line in response.message.get("response", []):
        file_name = response_line.get("file_name")
        review = response_line.get("review", [])
        log.debug("Review for file: %s", file_name)
        for review_line in review:
            line_number = review_line.get("line_number")
            line_code = review_line.get("line_code")
            suggestion = review_line.get("suggestion")
            log.debug("%s | %s | %s", line_number, line_code, suggestion)
            insert_comment(line_number, suggestion, file_name, pull_request, ai_user)


def insert_comment(line_number: int, text: str, f_path: str, pull_request: PullRequest, ai_user: User):
    target_repo: Repository = pull_request.target_repo
    repo_id = target_repo.repo_id
    line_number_formatted = "n%s" % line_number

    comment_type = "note"

    CommentsModel().create(
        text=text,
        repo=repo_id,
        user=ai_user.user_id,
        pull_request=pull_request,
        f_path=f_path,
        line_no=line_number_formatted,
        comment_type=comment_type,
        send_email=False,
    )


def _validate_settings(rc_settings):
    message = None
    valid = True

    if not rc_settings.get_setting_by_name("ai_api_key"):
        message = "AI API key is not set, review not possible."
        valid = False
    if not rc_settings.get_setting_by_name("ai_model"):
        message = "AI model is not set, review not possible."
        valid = False
    if not rc_settings.get_setting_by_name("ai_model_version"):
        message = "AI model version is not set, review not possible."
        valid = False

    return valid, message


def _get_diffset(
    pull_request: PullRequest,
):
    source_repo: Repository = pull_request.source_repo
    target_repo: Repository = pull_request.target_repo

    source_ref_id = pull_request.source_ref_parts.commit_id
    target_ref_id = pull_request.target_ref_parts.commit_id

    pre_load = ["author", "date", "message", "branch", "parents"]
    target_commit_final = target_repo.get_commit(commit_id=safe_str(target_ref_id), pre_load=pre_load)
    source_commit_final = source_repo.get_commit(commit_id=safe_str(source_ref_id), pre_load=pre_load)

    vcs_diff = PullRequestModel().get_diff(source_repo, source_ref_id, target_ref_id, False, MAX_CONTEXT)

    diff_processor = diffs.DiffProcessor(vcs_diff, diff_format="newdiff")

    _parsed = diff_processor.prepare()

    diffset = codeblocks.DiffSet(
        repo_name=target_repo.repo_name,
        source_repo_name=source_repo.repo_name,
        source_node_getter=codeblocks.diffset_node_getter(target_commit_final),
        target_node_getter=codeblocks.diffset_node_getter(source_commit_final),
    )

    return diffset.render_patchset(_parsed, source_ref=source_ref_id, target_ref=target_ref_id)


@async_task
@adopt_for_celery
def repo_size(extras):
    from rhodecode.lib.hooks_base import repo_size

    return repo_size(extras)


@async_task
@adopt_for_celery
def pre_pull(extras):
    from rhodecode.lib.hooks_base import pre_pull

    return pre_pull(extras)


@async_task
@adopt_for_celery
def post_pull(extras):
    from rhodecode.lib.hooks_base import post_pull

    return post_pull(extras)


@async_task
@adopt_for_celery
def pre_push(extras):
    from rhodecode.lib.hooks_base import pre_push

    return pre_push(extras)


@async_task
@adopt_for_celery
def post_push(extras):
    from rhodecode.lib.hooks_base import post_push

    return post_push(extras)