Ruff: Add PLC0415

dojo/api_v2/views.py

@@ -40,6 +40,7 @@
     prefetch,
     serializers,
 )
+from dojo.api_v2.prefetch.prefetcher import _Prefetcher
 from dojo.authorization.roles_permissions import Permissions
 from dojo.cred.queries import get_authorized_cred_mappings
 from dojo.endpoint.queries import (
@@ -584,8 +585,6 @@ def files(self, request, pk=None):
     )
     @action(detail=True, methods=["get", "post"])
     def complete_checklist(self, request, pk=None):
-        from dojo.api_v2.prefetch.prefetcher import _Prefetcher
-
         engagement = self.get_object()
         check_lists = Check_List.objects.filter(engagement=engagement)
         if request.method == "POST":

dojo/apps.py

@@ -71,21 +71,21 @@ def ready(self):
 
         # Load any signals here that will be ready for runtime
         # Importing the signals file is good enough if using the reciever decorator
-        import dojo.announcement.signals
-        import dojo.benchmark.signals
-        import dojo.cred.signals
-        import dojo.endpoint.signals
-        import dojo.engagement.signals
-        import dojo.file_uploads.signals
-        import dojo.finding_group.signals
-        import dojo.notes.signals
-        import dojo.product.signals
-        import dojo.product_type.signals
-        import dojo.risk_acceptance.signals
-        import dojo.sla_config.helpers
-        import dojo.tags_signals
-        import dojo.test.signals
-        import dojo.tool_product.signals  # noqa: F401
+        import dojo.announcement.signals  # noqa: PLC0415 raised: AppRegistryNotReady
+        import dojo.benchmark.signals  # noqa: PLC0415 raised: AppRegistryNotReady
+        import dojo.cred.signals  # noqa: PLC0415 raised: AppRegistryNotReady
+        import dojo.endpoint.signals  # noqa: PLC0415 raised: AppRegistryNotReady
+        import dojo.engagement.signals  # noqa: PLC0415 raised: AppRegistryNotReady
+        import dojo.file_uploads.signals  # noqa: PLC0415 raised: AppRegistryNotReady
+        import dojo.finding_group.signals  # noqa: PLC0415 raised: AppRegistryNotReady
+        import dojo.notes.signals  # noqa: PLC0415 raised: AppRegistryNotReady
+        import dojo.product.signals  # noqa: PLC0415 raised: AppRegistryNotReady
+        import dojo.product_type.signals  # noqa: PLC0415 raised: AppRegistryNotReady
+        import dojo.risk_acceptance.signals  # noqa: PLC0415 raised: AppRegistryNotReady
+        import dojo.sla_config.helpers  # noqa: PLC0415 raised: AppRegistryNotReady
+        import dojo.tags_signals  # noqa: PLC0415 raised: AppRegistryNotReady
+        import dojo.test.signals  # noqa: PLC0415 raised: AppRegistryNotReady
+        import dojo.tool_product.signals  # noqa: F401,PLC0415 raised: AppRegistryNotReady
 
 
 def get_model_fields_with_extra(model, extra_fields=()):

dojo/celery.py

@@ -1,5 +1,6 @@
 import logging
 import os
+from logging.config import dictConfig
 
 from celery import Celery
 from celery.signals import setup_logging
@@ -26,7 +27,6 @@ def debug_task(self):
 
 @setup_logging.connect
 def config_loggers(*args, **kwags):
-    from logging.config import dictConfig
     dictConfig(settings.LOGGING)
 
 

dojo/context_processors.py

@@ -1,8 +1,11 @@
 import contextlib
+import time
 
 # import the settings file
 from django.conf import settings
 
+from dojo.models import Alerts, System_Settings, UserAnnouncement
+
 
 def globalize_vars(request):
     # return the value you want as a dictionnary. you may add multiple values in there.
@@ -34,23 +37,18 @@ def globalize_vars(request):
 
 
 def bind_system_settings(request):
-    from dojo.models import System_Settings
-
     return {"system_settings": System_Settings.objects.get()}
 
 
 def bind_alert_count(request):
     if not settings.DISABLE_ALERT_COUNTER:
-        from dojo.models import Alerts
 
         if hasattr(request, "user") and request.user.is_authenticated:
             return {"alert_count": Alerts.objects.filter(user_id=request.user).count()}
     return {}
 
 
 def bind_announcement(request):
-    from dojo.models import UserAnnouncement
-
     with contextlib.suppress(Exception):  # TODO: this should be replaced with more meaningful exception
         if request.user.is_authenticated:
             user_announcement = UserAnnouncement.objects.select_related(
@@ -61,8 +59,6 @@ def bind_announcement(request):
 
 
 def session_expiry_notification(request):
-    import time
-
     try:
         if request.user.is_authenticated:
             last_activity = request.session.get("_last_activity", time.time())

dojo/decorators.py

@@ -55,8 +55,7 @@ def get_tasks(self):
 
 
 def we_want_async(*args, func=None, **kwargs):
-    from dojo.models import Dojo_User
-    from dojo.utils import get_current_user
+    from dojo.utils import get_current_user  # noqa: PLC0415 circular import
 
     sync = kwargs.get("sync", False)
     if sync:
@@ -83,7 +82,7 @@ def we_want_async(*args, func=None, **kwargs):
 def dojo_async_task(func):
     @wraps(func)
     def __wrapper__(*args, **kwargs):
-        from dojo.utils import get_current_user
+        from dojo.utils import get_current_user  # noqa: PLC0415 circular import
         user = get_current_user()
         kwargs["async_user"] = user
 

dojo/filters.py

@@ -3465,7 +3465,6 @@ class Meta:
 
 
 class LogEntryFilter(DojoFilter):
-    from auditlog.models import LogEntry
 
     action = MultipleChoiceFilter(choices=LogEntry.Action.choices)
     actor = ModelMultipleChoiceFilter(queryset=Dojo_User.objects.none())

dojo/finding/helper.py

@@ -27,7 +27,15 @@
     Vulnerability_Id_Template,
 )
 from dojo.notes.helper import delete_related_notes
-from dojo.utils import get_current_user, mass_model_updater, to_str_typed
+from dojo.tools import tool_issue_updater
+from dojo.utils import (
+    calculate_grade,
+    do_dedupe_finding,
+    do_false_positive_history,
+    get_current_user,
+    mass_model_updater,
+    to_str_typed,
+)
 
 logger = logging.getLogger(__name__)
 deduplicationLogger = logging.getLogger("dojo.specific-loggers.deduplication")
@@ -366,7 +374,6 @@ def post_process_finding_save(finding, dedupe_option=True, rules_option=True, pr
     if dedupe_option:
         if finding.hash_code is not None:
             if system_settings.enable_deduplication:
-                from dojo.utils import do_dedupe_finding
                 do_dedupe_finding(finding, *args, **kwargs)
             else:
                 deduplicationLogger.debug("skipping dedupe because it's disabled in system settings")
@@ -378,25 +385,21 @@ def post_process_finding_save(finding, dedupe_option=True, rules_option=True, pr
         if system_settings.enable_deduplication:
             deduplicationLogger.warning("skipping false positive history because deduplication is also enabled")
         else:
-            from dojo.utils import do_false_positive_history
             do_false_positive_history(finding, *args, **kwargs)
 
     # STEP 2 run all non-status changing tasks as celery tasks in the background
     if issue_updater_option:
-        from dojo.tools import tool_issue_updater
         tool_issue_updater.async_tool_issue_update(finding)
 
     if product_grading_option:
         if system_settings.enable_product_grade:
-            from dojo.utils import calculate_grade
             calculate_grade(finding.test.engagement.product)
         else:
             deduplicationLogger.debug("skipping product grading because it's disabled in system settings")
 
     # Adding a snippet here for push to JIRA so that it's in one place
     if push_to_jira:
         logger.debug("pushing finding %s to jira from finding.save()", finding.pk)
-        import dojo.jira_link.helper as jira_helper
 
         # current approach is that whenever a finding is in a group, the group will be pushed to JIRA
         # based on feedback we could introduct another push_group_to_jira boolean everywhere

dojo/finding/views.py

@@ -101,6 +101,7 @@
 )
 from dojo.notifications.helper import create_notification
 from dojo.test.queries import get_authorized_tests
+from dojo.tools import tool_issue_updater
 from dojo.utils import (
     FileIterWrapper,
     Product_Tab,
@@ -2918,8 +2919,6 @@ def finding_bulk_update_all(request, pid=None):
             error_counts = defaultdict(lambda: 0)
             success_count = 0
             for finding in finds:
-                from dojo.tools import tool_issue_updater
-
                 tool_issue_updater.async_tool_issue_update(finding)
 
                 # not sure yet if we want to support bulk unlink, so leave as commented out for now

dojo/forms.py

@@ -18,6 +18,7 @@
 from django.contrib.auth.password_validation import validate_password
 from django.core import validators
 from django.core.exceptions import ValidationError
+from django.core.validators import URLValidator
 from django.db.models import Count, Q
 from django.forms import modelformset_factory
 from django.forms.widgets import Select, Widget
@@ -384,8 +385,6 @@ class EditFindingGroupForm(forms.ModelForm):
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        import dojo.jira_link.helper as jira_helper
-
         self.fields["push_to_jira"] = forms.BooleanField()
         self.fields["push_to_jira"].required = False
         self.fields["push_to_jira"].help_text = "Checking this will overwrite content of your JIRA issue, or create one."
@@ -2596,7 +2595,6 @@ class BaseJiraForm(forms.ModelForm):
     password = forms.CharField(widget=forms.PasswordInput, required=True, help_text=JIRA_Instance._meta.get_field("password").help_text, label=JIRA_Instance._meta.get_field("password").verbose_name)
 
     def test_jira_connection(self):
-        import dojo.jira_link.helper as jira_helper
         try:
             # Attempt to validate the credentials before moving forward
             jira_helper.get_jira_connection_raw(self.cleaned_data["url"],
@@ -2665,13 +2663,6 @@ class Meta:
         fields = ["id"]
 
 
-# class JIRA_ProjectForm(forms.ModelForm):
-
-#     class Meta:
-#         model = JIRA_Project
-#         exclude = ['product']
-
-
 class Product_API_Scan_ConfigurationForm(forms.ModelForm):
 
     def __init__(self, *args, **kwargs):
@@ -2768,7 +2759,6 @@ class Meta:
         exclude = ["product"]
 
     def clean(self):
-        from django.core.validators import URLValidator
         form_data = self.cleaned_data
 
         try:
@@ -2849,7 +2839,6 @@ class Meta:
         order = ["name"]
 
     def clean(self):
-        from django.core.validators import URLValidator
         form_data = self.cleaned_data
 
         try:
@@ -3061,7 +3050,6 @@ class Meta:
         fields = ["inherit_from_product", "jira_instance", "project_key", "issue_template_dir", "epic_issue_type_name", "component", "custom_fields", "jira_labels", "default_assignee", "enabled", "add_vulnerability_id_to_jira_label", "push_all_issues", "enable_engagement_epic_mapping", "push_notes", "product_jira_sla_notification", "risk_acceptance_expiration_notification"]
 
     def __init__(self, *args, **kwargs):
-        from dojo.jira_link import helper as jira_helper
         # if the form is shown for an engagement, we set a placeholder text around inherited settings from product
         self.target = kwargs.pop("target", "product")
         self.product = kwargs.pop("product", None)

dojo/github.py

@@ -16,7 +16,7 @@
 
 def reopen_external_issue_github(find, note, prod, eng):
 
-    from dojo.utils import get_system_setting
+    from dojo.utils import get_system_setting  # noqa: PLC0415 circular import
     if not get_system_setting("enable_github"):
         return
 
@@ -47,7 +47,7 @@ def reopen_external_issue_github(find, note, prod, eng):
 
 def close_external_issue_github(find, note, prod, eng):
 
-    from dojo.utils import get_system_setting
+    from dojo.utils import get_system_setting  # noqa: PLC0415 circular import
     if not get_system_setting("enable_github"):
         return
 
@@ -78,7 +78,7 @@ def close_external_issue_github(find, note, prod, eng):
 
 def update_external_issue_github(find, prod, eng):
 
-    from dojo.utils import get_system_setting
+    from dojo.utils import get_system_setting  # noqa: PLC0415 circular import
     if not get_system_setting("enable_github"):
         return
 
@@ -106,7 +106,7 @@ def update_external_issue_github(find, prod, eng):
 
 def add_external_issue_github(find, prod, eng):
 
-    from dojo.utils import get_system_setting
+    from dojo.utils import get_system_setting  # noqa: PLC0415 circular import
     if not get_system_setting("enable_github"):
         return
 

dojo/jira_link/helper.py

@@ -1,3 +1,4 @@
+import importlib
 import io
 import json
 import logging
@@ -442,7 +443,6 @@ def connect_to_jira(jira_server, jira_username, jira_password):
 def get_jira_connect_method():
     if hasattr(settings, "JIRA_CONNECT_METHOD"):
         try:
-            import importlib
             mn, _, fn = settings.JIRA_CONNECT_METHOD.rpartition(".")
             m = importlib.import_module(mn)
             return getattr(m, fn)
@@ -1770,7 +1770,7 @@ def escape_for_jira(text):
 
 def process_resolution_from_jira(finding, resolution_id, resolution_name, assignee_name, jira_now, jira_issue, finding_group: Finding_Group = None) -> bool:
     """Processes the resolution field in the JIRA issue and updated the finding in Defect Dojo accordingly"""
-    import dojo.risk_acceptance.helper as ra_helper
+    import dojo.risk_acceptance.helper as ra_helper  # noqa: PLC0415 import error
     status_changed = False
     resolved = resolution_id is not None
     jira_instance = get_jira_instance(finding)

dojo/middleware.py

@@ -13,6 +13,7 @@
 from django.urls import reverse
 from django.utils.functional import SimpleLazyObject
 
+from dojo.models import Dojo_User
 from dojo.product_announcements import LongRunningRequestProductAnnouncement
 
 logger = logging.getLogger(__name__)
@@ -65,7 +66,6 @@ def __call__(self, request):
                 # this populates dd_user log var, so can appear in the uwsgi logs
                 uwsgi.set_logvar("dd_user", str(request.user))
             path = request.path_info.lstrip("/")
-            from dojo.models import Dojo_User
             if Dojo_User.force_password_reset(request.user) and path != "change_password":
                 return HttpResponseRedirect(reverse("change_password"))
 
@@ -77,8 +77,7 @@ class DojoSytemSettingsMiddleware:
 
     def __init__(self, get_response):
         self.get_response = get_response
-        # avoid circular imports
-        from dojo.models import System_Settings
+        from dojo.models import System_Settings  # noqa: PLC0415 circular import
         models.signals.post_save.connect(self.cleanup, sender=System_Settings)
 
     def __call__(self, request):
@@ -107,7 +106,7 @@ def cleanup(cls, *args, **kwargs):  # noqa: ARG003
     def load(cls):
         # cleanup any existing settings first to ensure fresh state
         cls.cleanup()
-        from dojo.models import System_Settings
+        from dojo.models import System_Settings  # noqa: PLC0415 circular import
         system_settings = System_Settings.objects.get(no_cache=True)
         cls._thread_local.system_settings = system_settings
         return system_settings
@@ -120,7 +119,7 @@ def get_from_db(self, *args, **kwargs):
         try:
             from_db = super().get(*args, **kwargs)
         except:
-            from dojo.models import System_Settings
+            from dojo.models import System_Settings  # noqa: PLC0415 circular import
             # this mimics the existing code that was in filters.py and utils.py.
             # cases I have seen triggering this is for example manage.py collectstatic inside a docker build where mysql is not available
             # logger.debug('unable to get system_settings from database, constructing (new) default instance. Exception was:', exc_info=True)

dojo/notifications/helper.py

@@ -696,8 +696,7 @@ def _process_objects(self, **kwargs: dict) -> None:
             self.product = finding.test.engagement.product
             logger.debug("Defined product of finding %s", self.product)
         elif (obj := kwargs.get("obj")) is not None:
-            from dojo.utils import get_product
-
+            from dojo.utils import get_product  # noqa: PLC0415 circular import
             self.product = get_product(obj)
             logger.debug("Defined product of obj %s", self.product)