25400571889

Committed 05 May 2026 08:29PM UTC coverage: 90.612% (-0.4%) from 90.986%

Build # 25400571889

Build Type

Pull #643

github

Committed by

web-flow

Commit Message

Merge 4e812ec20 into 15d676845

Pull Request Pull Request #643: Adding new backfill option

Coverage Stats

8291 of 9150 relevant lines covered (90.61%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.44

back/admin/integrations/views.py

import json
from datetime import timedelta
from urllib.parse import urlparse

import requests
from django.contrib import messages
from django.contrib.messages.views import SuccessMessageMixin
from django.http import Http404, HttpResponseRedirect
from django.shortcuts import get_object_or_404, redirect
from django.urls import reverse_lazy
from django.utils import timezone
from django.utils.translation import gettext as _
from django.views.generic import View
from django.views.generic.base import RedirectView
from django.views.generic.detail import DetailView
from django.views.generic.edit import CreateView, DeleteView, UpdateView
from django.views.generic.list import ListView
from django_q.tasks import async_task

from users.mixins import AdminOrManagerPermMixin, AdminPermMixin

from .forms import IntegrationExtraArgsForm, IntegrationForm
from .models import Integration, IntegrationTracker


class IntegrationCreateView(AdminPermMixin, CreateView, SuccessMessageMixin):
    template_name = "token_create.html"
    form_class = IntegrationForm
    success_message = _("Integration has been created!")
    success_url = reverse_lazy("settings:integrations")

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["title"] = _("Add new integration")
        context["subtitle"] = _("settings")
        context["button_text"] = _("Create")
        return context

    def form_valid(self, form):
        form.instance.integration = Integration.Type.CUSTOM
        return super().form_valid(form)


class IntegrationUpdateView(AdminPermMixin, UpdateView, SuccessMessageMixin):
    template_name = "token_create.html"
    form_class = IntegrationForm
    queryset = Integration.objects.filter(integration=Integration.Type.CUSTOM)
    success_message = _("Integration has been updated!")
    success_url = reverse_lazy("settings:integrations")

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["title"] = _("Update existing integration")
        context["subtitle"] = _("settings")
        context["button_text"] = _("Update")
        return context

    def form_valid(self, form):
        new_initial_data = form.cleaned_data["manifest"].get("initial_data_form", [])
        old_initial_data = self.get_object().manifest.get("initial_data_form", [])

        # remove keys that don't exist anymore from saved secrets
        new_initial_data_keys = [item["id"] for item in new_initial_data]
        for item in old_initial_data:
            if item["id"] not in new_initial_data_keys:
                form.instance.extra_args.pop(item["id"], None)

        return super().form_valid(form)


class IntegrationDeleteView(AdminPermMixin, DeleteView):
    """This is a general delete function for all integrations"""

    template_name = "integration-delete.html"
    model = Integration
    success_url = reverse_lazy("settings:integrations")

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["title"] = _("Delete integration")
        context["subtitle"] = _("settings")
        return context


class IntegrationUpdateExtraArgsView(AdminPermMixin, UpdateView, SuccessMessageMixin):
    template_name = "update_initial_data_form.html"
    form_class = IntegrationExtraArgsForm
    queryset = Integration.objects.filter(integration=Integration.Type.CUSTOM)
    success_message = _("Your config values have been updated!")
    success_url = reverse_lazy("settings:integrations")

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["title"] = _("Integration settings")
        context["subtitle"] = _("settings")
        context["button_text"] = _("Update")
        return context


class IntegrationDeleteExtraArgsView(AdminPermMixin, DeleteView, SuccessMessageMixin):
    template_name = "update_initial_data_form.html"
    queryset = Integration.objects.filter(integration=Integration.Type.CUSTOM)
    success_message = _("Secret value has been removed")
    success_url = reverse_lazy("settings:integrations")

    def form_valid(self, form):
        self.object = self.get_object()

        secret_value = self.kwargs.get("secret")
        if secret_value not in [
            item["id"] for item in self.object.filled_secret_values
        ]:
            raise Http404

        self.object.extra_args.pop(secret_value)
        self.object.save()
        success_url = reverse_lazy("integrations:update-creds", args=[self.object.pk])
        return HttpResponseRedirect(success_url)


class IntegrationOauthRedirectView(RedirectView):
    permanent = False

    def get_redirect_url(self, pk, *args, **kwargs):
        integration = get_object_or_404(
            Integration,
            pk=pk,
            manifest__oauth__isnull=False,
            enabled_oauth=False,
        )
        return integration._replace_vars(
            integration.manifest["oauth"]["authenticate_url"]
        )


class IntegrationOauthCallbackView(RedirectView):
    permanent = False

    def get_redirect_url(self, pk, *args, **kwargs):
        integration = get_object_or_404(
            Integration,
            pk=pk,
            manifest__oauth__isnull=False,
            enabled_oauth=False,
        )
        code = self.request.GET.get("code", "")
        if code == "" and not integration.manifest["oauth"].get("without_code", False):
            messages.error(self.request, "Code was not provided")
            return reverse_lazy("settings:integrations")

        # Check if url has parameters already
        access_obj = integration.manifest["oauth"]["access_token"]
        if not integration.manifest["oauth"].get("without_code", False):
            parsed_url = urlparse(access_obj["url"])
            if len(parsed_url.query):
                access_obj["url"] += "&code=" + code
            else:
                access_obj["url"] += "?code=" + code

        success, response = integration.run_request(access_obj)

        if not success:
            messages.error(self.request, f"Couldn't save token: {response}")
            return reverse_lazy("settings:integrations")

        integration.extra_args["oauth"] = response.json()
        if "expires_in" in response.json():
            integration.expiring = timezone.now() + timedelta(
                seconds=response.json()["expires_in"]
            )

        integration.enabled_oauth = True
        integration.save(update_fields=["enabled_oauth", "extra_args", "expiring"])

        return reverse_lazy("settings:integrations")


class SlackOAuthView(View):
    def get(self, request):
        access_token, _dummy = Integration.objects.get_or_create(
            integration=Integration.Type.SLACK_BOT
        )
        if "code" not in request.GET:
            messages.error(
                request,
                _("Could not optain slack authentication code."),
            )
            return redirect("settings:integrations")
        code = request.GET["code"]
        params = {
            "code": code,
            "client_id": access_token.client_id,
            "client_secret": access_token.client_secret,
            "redirect_uri": access_token.redirect_url,
        }
        url = "https://slack.com/api/oauth.v2.access"
        json_response = requests.get(url, params)
        data = json.loads(json_response.text)
        if data["ok"]:
            access_token.bot_token = data["access_token"]
            access_token.bot_id = data["bot_user_id"]
            access_token.token = data["access_token"]
            access_token.save()
            messages.success(
                request,
                _(
                    "Slack has successfully been connected. You have a new bot in your "
                    "workspace."
                ),
            )
        else:
            messages.error(request, _("Could not get tokens from Slack"))
        return redirect("settings:integrations")


class IntegrationTrackerListView(AdminOrManagerPermMixin, ListView):
    queryset = (
        IntegrationTracker.objects.all()
        .select_related("integration", "for_user")
        .filter(integration__is_active=True)
        .order_by("-ran_at")
    )
    template_name = "tracker_list.html"

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["title"] = _("All integration runs")
        context["subtitle"] = _("integrations")
        return context


class IntegrationTrackerDetailView(AdminOrManagerPermMixin, DetailView):
    model = IntegrationTracker
    template_name = "tracker.html"

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["title"] = _("%(integration)s for %(user)s") % {
            "integration": (
                self.object.integration.name
                if self.object.integration is not None
                else "Test integration"
            ),
            "user": self.object.for_user,
        }
        context["subtitle"] = _("integrations")
        return context


class IntegrationBackfillIDsView(AdminPermMixin, View):
    def post(self, request, pk):
        integration = get_object_or_404(Integration, pk=pk)
        if not integration.can_backfill_ids:
            messages.error(
                request,
                _("This integration has no store_data declared on its exists block."),
            )
            return redirect("settings:integrations")
        async_task(
            "admin.integrations.tasks.backfill_integration_ids",
            integration.id,
            task_name=f"Backfill IDs: {integration.name}",
        )
        messages.success(
            request,
            _("Backfill started for %(name)s. Users' extra fields will populate "
              "as the lookup runs in the background.") % {"name": integration.name},
        )
        return redirect("settings:integrations")

1	import json	1✔
2	from datetime import timedelta	1✔
3	from urllib.parse import urlparse	1✔
4
5	import requests	1✔
6	from django.contrib import messages	1✔
7	from django.contrib.messages.views import SuccessMessageMixin	1✔
8	from django.http import Http404, HttpResponseRedirect	1✔
9	from django.shortcuts import get_object_or_404, redirect	1✔
10	from django.urls import reverse_lazy	1✔
11	from django.utils import timezone	1✔
12	from django.utils.translation import gettext as _	1✔
13	from django.views.generic import View	1✔
14	from django.views.generic.base import RedirectView	1✔
15	from django.views.generic.detail import DetailView	1✔
16	from django.views.generic.edit import CreateView, DeleteView, UpdateView	1✔
17	from django.views.generic.list import ListView	1✔
18	from django_q.tasks import async_task	1✔
19
20	from users.mixins import AdminOrManagerPermMixin, AdminPermMixin	1✔
21
22	from .forms import IntegrationExtraArgsForm, IntegrationForm	1✔
23	from .models import Integration, IntegrationTracker	1✔
24
25
26	class IntegrationCreateView(AdminPermMixin, CreateView, SuccessMessageMixin):	1✔
27	template_name = "token_create.html"	1✔
28	form_class = IntegrationForm	1✔
29	success_message = _("Integration has been created!")	1✔
30	success_url = reverse_lazy("settings:integrations")	1✔
31
32	def get_context_data(self, **kwargs):	1✔
33	context = super().get_context_data(**kwargs)	1✔
34	context["title"] = _("Add new integration")	1✔
35	context["subtitle"] = _("settings")	1✔
36	context["button_text"] = _("Create")	1✔
37	return context	1✔
38
39	def form_valid(self, form):	1✔
40	form.instance.integration = Integration.Type.CUSTOM	1✔
41	return super().form_valid(form)	1✔
42
43
44	class IntegrationUpdateView(AdminPermMixin, UpdateView, SuccessMessageMixin):	1✔
45	template_name = "token_create.html"	1✔
46	form_class = IntegrationForm	1✔
47	queryset = Integration.objects.filter(integration=Integration.Type.CUSTOM)	1✔
48	success_message = _("Integration has been updated!")	1✔
49	success_url = reverse_lazy("settings:integrations")	1✔
50
51	def get_context_data(self, **kwargs):	1✔
52	context = super().get_context_data(**kwargs)	1✔
53	context["title"] = _("Update existing integration")	1✔
54	context["subtitle"] = _("settings")	1✔
55	context["button_text"] = _("Update")	1✔
56	return context	1✔
57
58	def form_valid(self, form):	1✔
59	new_initial_data = form.cleaned_data["manifest"].get("initial_data_form", [])	1✔
60	old_initial_data = self.get_object().manifest.get("initial_data_form", [])	1✔
61
62	# remove keys that don't exist anymore from saved secrets
63	new_initial_data_keys = [item["id"] for item in new_initial_data]	1✔
64	for item in old_initial_data:	1✔
65	if item["id"] not in new_initial_data_keys:	1✔
66	form.instance.extra_args.pop(item["id"], None)	1✔
67
68	return super().form_valid(form)	1✔
69
70
71	class IntegrationDeleteView(AdminPermMixin, DeleteView):	1✔
72	"""This is a general delete function for all integrations"""
73
74	template_name = "integration-delete.html"	1✔
75	model = Integration	1✔
76	success_url = reverse_lazy("settings:integrations")	1✔
77
78	def get_context_data(self, **kwargs):	1✔
79	context = super().get_context_data(**kwargs)	1✔
80	context["title"] = _("Delete integration")	1✔
81	context["subtitle"] = _("settings")	1✔
82	return context	1✔
83
84
85	class IntegrationUpdateExtraArgsView(AdminPermMixin, UpdateView, SuccessMessageMixin):	1✔
86	template_name = "update_initial_data_form.html"	1✔
87	form_class = IntegrationExtraArgsForm	1✔
88	queryset = Integration.objects.filter(integration=Integration.Type.CUSTOM)	1✔
89	success_message = _("Your config values have been updated!")	1✔
90	success_url = reverse_lazy("settings:integrations")	1✔
91
92	def get_context_data(self, **kwargs):	1✔
93	context = super().get_context_data(**kwargs)	1✔
94	context["title"] = _("Integration settings")	1✔
95	context["subtitle"] = _("settings")	1✔
96	context["button_text"] = _("Update")	1✔
97	return context	1✔
98
99
100	class IntegrationDeleteExtraArgsView(AdminPermMixin, DeleteView, SuccessMessageMixin):	1✔
101	template_name = "update_initial_data_form.html"	1✔
102	queryset = Integration.objects.filter(integration=Integration.Type.CUSTOM)	1✔
103	success_message = _("Secret value has been removed")	1✔
104	success_url = reverse_lazy("settings:integrations")	1✔
105
106	def form_valid(self, form):	1✔
107	self.object = self.get_object()	1✔
108
109	secret_value = self.kwargs.get("secret")	1✔
110	if secret_value not in [	1✔
111	item["id"] for item in self.object.filled_secret_values
112	]:
113	raise Http404	1✔
114
115	self.object.extra_args.pop(secret_value)	1✔
116	self.object.save()	1✔
117	success_url = reverse_lazy("integrations:update-creds", args=[self.object.pk])	1✔
118	return HttpResponseRedirect(success_url)	1✔
119
120
121	class IntegrationOauthRedirectView(RedirectView):	1✔
122	permanent = False	1✔
123
124	def get_redirect_url(self, pk, args, *kwargs):	1✔
125	integration = get_object_or_404(	1✔
126	Integration,
127	pk=pk,
128	manifest__oauth__isnull=False,
129	enabled_oauth=False,
130	)
131	return integration._replace_vars(	1✔
132	integration.manifest["oauth"]["authenticate_url"]
133	)
134
135
136	class IntegrationOauthCallbackView(RedirectView):	1✔
137	permanent = False	1✔
138
139	def get_redirect_url(self, pk, args, *kwargs):	1✔
140	integration = get_object_or_404(	1✔
141	Integration,
142	pk=pk,
143	manifest__oauth__isnull=False,
144	enabled_oauth=False,
145	)
146	code = self.request.GET.get("code", "")	1✔
147	if code == "" and not integration.manifest["oauth"].get("without_code", False):	1✔
148	messages.error(self.request, "Code was not provided")	1✔
149	return reverse_lazy("settings:integrations")	1✔
150
151	# Check if url has parameters already
152	access_obj = integration.manifest["oauth"]["access_token"]	1✔
153	if not integration.manifest["oauth"].get("without_code", False):	1✔
154	parsed_url = urlparse(access_obj["url"])	1✔
155	if len(parsed_url.query):	1✔
156	access_obj["url"] += "&code=" + code	1✔
157	else:
158	access_obj["url"] += "?code=" + code	1✔
159
160	success, response = integration.run_request(access_obj)	1✔
161
162	if not success:	1✔
163	messages.error(self.request, f"Couldn't save token: {response}")	1✔
164	return reverse_lazy("settings:integrations")	1✔
165
166	integration.extra_args["oauth"] = response.json()	1✔
167	if "expires_in" in response.json():	1✔
168	integration.expiring = timezone.now() + timedelta(	×
169	seconds=response.json()["expires_in"]
170	)
171
172	integration.enabled_oauth = True	1✔
173	integration.save(update_fields=["enabled_oauth", "extra_args", "expiring"])	1✔
174
175	return reverse_lazy("settings:integrations")	1✔
176
177
178	class SlackOAuthView(View):	1✔
179	def get(self, request):	1✔
180	access_token, _dummy = Integration.objects.get_or_create(	1✔
181	integration=Integration.Type.SLACK_BOT
182	)
183	if "code" not in request.GET:	1✔
184	messages.error(	1✔
185	request,
186	_("Could not optain slack authentication code."),
187	)
188	return redirect("settings:integrations")	1✔
189	code = request.GET["code"]	1✔
190	params = {	1✔
191	"code": code,
192	"client_id": access_token.client_id,
193	"client_secret": access_token.client_secret,
194	"redirect_uri": access_token.redirect_url,
195	}
196	url = "https://slack.com/api/oauth.v2.access"	1✔
197	json_response = requests.get(url, params)	1✔
198	data = json.loads(json_response.text)	1✔
199	if data["ok"]:	1✔
200	access_token.bot_token = data["access_token"]	1✔
201	access_token.bot_id = data["bot_user_id"]	1✔
202	access_token.token = data["access_token"]	1✔
203	access_token.save()	1✔
204	messages.success(	1✔
205	request,
206	_(
207	"Slack has successfully been connected. You have a new bot in your "
208	"workspace."
209	),
210	)
211	else:
212	messages.error(request, _("Could not get tokens from Slack"))	×
213	return redirect("settings:integrations")	1✔
214
215
216	class IntegrationTrackerListView(AdminOrManagerPermMixin, ListView):	1✔
217	queryset = (	1✔
218	IntegrationTracker.objects.all()
219	.select_related("integration", "for_user")
220	.filter(integration__is_active=True)
221	.order_by("-ran_at")
222	)
223	template_name = "tracker_list.html"	1✔
224
225	def get_context_data(self, **kwargs):	1✔
226	context = super().get_context_data(**kwargs)	1✔
227	context["title"] = _("All integration runs")	1✔
228	context["subtitle"] = _("integrations")	1✔
229	return context	1✔
230
231
232	class IntegrationTrackerDetailView(AdminOrManagerPermMixin, DetailView):	1✔
233	model = IntegrationTracker	1✔
234	template_name = "tracker.html"	1✔
235
236	def get_context_data(self, **kwargs):	1✔
237	context = super().get_context_data(**kwargs)	1✔
238	context["title"] = _("%(integration)s for %(user)s") % {	1✔
239	"integration": (
240	self.object.integration.name
241	if self.object.integration is not None
242	else "Test integration"
243	),
244	"user": self.object.for_user,
245	}
246	context["subtitle"] = _("integrations")	1✔
247	return context	1✔
248
249
250	class IntegrationBackfillIDsView(AdminPermMixin, View):	1✔
251	def post(self, request, pk):	1✔
252	integration = get_object_or_404(Integration, pk=pk)	×
253	if not integration.can_backfill_ids:	×
254	messages.error(	×
255	request,
256	_("This integration has no store_data declared on its exists block."),
257	)
258	return redirect("settings:integrations")	×
259	async_task(	×
260	"admin.integrations.tasks.backfill_integration_ids",
261	integration.id,
262	task_name=f"Backfill IDs: {integration.name}",
263	)
264	messages.success(	×
265	request,
266	_("Backfill started for %(name)s. Users' extra fields will populate "
267	"as the lookup runs in the background.") % {"name": integration.name},
268	)
269	return redirect("settings:integrations")	×

chiefonboarding / ChiefOnboarding / 25400571889

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous