biosample.views.CreateUserView.form_valid() - Code Metrics - cnr-ibba/IMAGE-InjectTool - Measure and Improve Code Quality continuously with Scrutinizer

biosample.views.CreateUserView.form_valid() B
last analyzed 2023-06-12 09:55 UTC

↳ Parent: biosample.views

Complexity

Conditions

Size

Total Lines	51
Code Lines	24

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	24
dl	0
loc	51
rs	8.8373
c	0
b	0
f	0
cc	5
nop	2

How to fix Long Method


import os
import re
import json
import redis
import logging
import traceback

from decouple import AutoConfig

from django.conf import settings
from django.urls import reverse_lazy, reverse
from django.utils.crypto import get_random_string
from django.utils.http import is_safe_url
from django.shortcuts import redirect, get_object_or_404
from django.contrib.auth.mixins import LoginRequiredMixin
from django.views.generic import TemplateView
from django.views.generic.edit import FormView, CreateView, ModelFormMixin
from django.contrib import messages
from django.http import HttpResponseRedirect

from pyUSIrest.usi import User
from pyUSIrest.exceptions import USIConnectionError

from common.constants import WAITING
from common.helpers import send_mail_to_admins
from uid.models import Submission

from .forms import (
    GenerateTokenForm, RegisterUserForm, CreateUserForm, SubmitForm)
from .models import Account, ManagedTeam
from .helpers import get_auth, get_manager_auth
from .tasks import SplitSubmissionTask


# Get an instance of a logger
logger = logging.getLogger(__name__)

# while near
TOKEN_DURATION_THRESHOLD = 3600*4

# define a decouple config object
settings_dir = os.path.join(settings.BASE_DIR, 'image')
config = AutoConfig(search_path=settings_dir)


# In programming, a mixin is a class that provides functionality to be
# inherited, but isn’t meant for instantiation on its own. In programming
# languages with multiple inheritance, mixins can be used to add enhanced
# functionality and behavior to classes.
class AccountMixin(object):
    """A generic mixin associated with biosample.models. You need to costomize
    account_found and account_not_found in to do a custom redirect in case
    a manager account is found or not"""

    def account_not_found(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

    def account_found(self, request, *args, **kwargs):
        return super().dispatch(request, *args, **kwargs)

    def dispatch(self, request, *args, **kwargs):
        # get user from request and user model. This could be also Anonymous
        # user:however with metod decorator a login is required before dispatch
        # method is called
        user = self.request.user

        if hasattr(user, "biosample_account"):
            return self.account_found(request, *args, **kwargs)

        else:
            return self.account_not_found(request, *args, **kwargs)


class TokenMixin(AccountMixin):
    """Get common stuff for Token visualization. Redirect to AAP registration
    if no valid AAP credentials are found for request.user"""

    def get_initial(self):
        """
        Returns the initial data to use for forms on this view.
        """

        initial = super(TokenMixin, self).get_initial()
        initial['name'] = self.request.user.biosample_account.name

        return initial

    # override AccountMixin method
    def account_not_found(self, request, *args, **kwargs):
        """If a user has not an account, redirect to activation complete"""

        logger.warning("Error for user:%s: not managed" % self.request.user)

        messages.warning(
            request=self.request,
            message='You need to register a valid AAP profile',
            extra_tags="alert alert-dismissible alert-warning")

        return redirect('accounts:registration_activation_complete')


class RegisterMixin(AccountMixin):
    """If a biosample account is already registered, returns to dashboard"""

    # override AccountMixin method
    def account_found(self, request, *args, **kwargs):
        """If a user has been registered, redirect to dashboard"""

        logger.warning(
            "Error for user:%s: Already registered" % self.request.user)

        messages.warning(
            request=self.request,
            message='Your AAP profile is already registered',
            extra_tags="alert alert-dismissible alert-warning")

        return redirect('uid:dashboard')


class MyFormMixin(object):
    """Common stuff for token generation"""

    success_url_message = "Please set this variable"
    success_url = reverse_lazy("uid:dashboard")

    # add the request to the kwargs
    # https://chriskief.com/2012/12/18/django-modelform-formview-and-the-request-object/
    # this is needed to display messages (django.contronb) on pages
    def get_form_kwargs(self):
        kwargs = super(MyFormMixin, self).get_form_kwargs()
        kwargs['request'] = self.request
        return kwargs

    def get_success_url(self):
        """Override default function"""

        messages.success(
            request=self.request,
            message=self.success_url_message,
            extra_tags="alert alert-dismissible alert-success")

        return super(MyFormMixin, self).get_success_url()

    def form_invalid(self, form):
        messages.error(
            self.request,
            message="Please correct the errors below",
            extra_tags="alert alert-dismissible alert-danger")

        return super(MyFormMixin, self).form_invalid(form)

    def generate_token(self, form):
        """Generate token from form instance"""

        # operate on form data
        name = form.cleaned_data['name']
        password = form.cleaned_data['password']

        try:
            auth = get_auth(user=name, password=password)

        except USIConnectionError as e:
            # logger exception. With repr() the exception name is rendered
            logger.error(repr(e))

            # try to detect the error message
            pattern = re.compile(r"(\{.*\})")
            match = re.search(pattern, str(e))

            if match is None:
                # parse error message
                messages.error(
                    self.request,
                    "Unable to generate token: %s" % str(e),
                    extra_tags="alert alert-dismissible alert-danger")

            else:
                data = json.loads(match.groups()[0])
                messages.error(
                    self.request,
                    "Unable to generate token: %s" % data['message'],
                    extra_tags="alert alert-dismissible alert-danger")

            # cant't return form_invalid here, since i need to process auth
            return None

        else:
            logger.debug("Token generated for user:%s" % name)

            self.request.session['token'] = auth.token

            # return an auth object
            return auth


class GenerateTokenView(LoginRequiredMixin, TokenMixin, MyFormMixin, FormView):
    """Generate AAP token. If user is not registered, redirect to accounts
    registration_activation_complete through TokenMixin. If yes generate
    token through MyFormMixin"""

    template_name = 'biosample/generate_token.html'
    form_class = GenerateTokenForm
    success_url_message = 'Token generated!'

    def dispatch(self, request, *args, **kwargs):
        # try to read next link
        next_url = request.GET.get('next', None)

        # redirect to next url. is_safe_url: is a safe redirection
        # (i.e. it doesn't point to a different host and uses a safe scheme).
        if next_url and is_safe_url(next_url, allowed_hosts=None):
            logger.debug("Got %s as next_url" % next_url)
            self.success_url = next_url

        return super(
            GenerateTokenView, self).dispatch(request, *args, **kwargs)

    def form_valid(self, form):
        # This method is called when valid form data has been POSTed.
        # It should return an HttpResponse.

        # call MyFormMixin method and generate token. Check user/password
        if not self.generate_token(form):
            return self.form_invalid(form)

        return redirect(self.get_success_url())


class TokenView(LoginRequiredMixin, TokenMixin, TemplateView):
    """Visualize token details"""

    template_name = 'biosample/token.html'

    def get_context_data(self, **kwargs):
        # Call the base implementation first to get a context
        context = super(TokenView, self).get_context_data(**kwargs)

        # get user and team object
        context['name'] = self.request.user.biosample_account.name
        context['team'] = self.request.user.biosample_account.team

        try:
            # add content to context
            auth = get_auth(token=self.request.session['token'])

            if auth.is_expired():
                messages.error(
                    self.request,
                    "Your token is expired",
                    extra_tags="alert alert-dismissible alert-danger")

            context["auth"] = auth

        except KeyError as e:
            logger.error(
                "No valid token found for %s: %s" % (
                    self.request.user,
                    e))

            messages.error(
                self.request,
                "You haven't generated any token yet",
                extra_tags="alert alert-dismissible alert-danger")

        return context


class RegisterUserView(LoginRequiredMixin, RegisterMixin, MyFormMixin,
                       CreateView):
    """Register an already existent AAP account"""

    template_name = 'biosample/register_user.html'
    form_class = RegisterUserForm
    success_url_message = 'Account registered'

    def form_valid(self, form):
        # This method is called when valid form data has been POSTed.
        # It should return an HttpResponse.
        team = form.cleaned_data['team']

        # call AuthMixin method and generate token. Check user/password
        auth = self.generate_token(form)

        if not auth:
            return self.form_invalid(form)

        if team.name not in auth.get_domains():
            messages.error(
                self.request,
                "You don't belong to team: %s" % team,
                extra_tags="alert alert-dismissible alert-danger")

            # return invalid form
            return self.form_invalid(form)

        # add a user to object (comes from section not from form)
        self.object = form.save(commit=False)
        self.object.user = self.request.user
        self.object.save()

        # call to a specific function (which does an HttpResponseRedirect
        # to success_url)
        return super(ModelFormMixin, self).form_valid(form)


class CreateUserView(LoginRequiredMixin, RegisterMixin, MyFormMixin, FormView):
    """Create a new AAP account"""

    template_name = 'biosample/create_user.html'
    form_class = CreateUserForm
    success_url_message = "AAP Account created"

    def deal_with_errors(self, error_message, exception):
        """Add messages to view for encountered errors

        Args:
            error_message (str): An informative text
            exception (Exception): an exception instance
        """

        exception_message = "Message was: %s" % (exception)

        logger.error(error_message)
        logger.error(exception_message)

        messages.error(
            self.request,
            message=error_message,
            extra_tags="alert alert-dismissible alert-danger")

        messages.error(
            self.request,
            message=exception_message,
            extra_tags="alert alert-dismissible alert-danger")

        # get exception info
        einfo = traceback.format_exc()

        # send a mail to admin
        send_mail_to_admins(error_message, einfo)

    def get_form_kwargs(self):
        """Override get_form_kwargs"""

        kwargs = super(CreateUserView, self).get_form_kwargs()

        # create a new biosample user
        username = "image-%s" % (get_random_string(length=8))

        # add username to instance
        kwargs['username'] = username

        return kwargs

    # create an user or throw an exception
    def create_biosample_user(self, form, full_name, affiliation):
        """Create a biosample user

        Args:
            form (:py:class:`CreateUserForm`) an instantiated form object
            full_name (str): the user full name (Name + Surname)
            affiliation (str): the organization the user belongs to

        Returns:
            str: a biosamples user id
        """

        password = form.cleaned_data['password1']
        confirmPwd = form.cleaned_data['password2']

        # get user model associated with this session
        user = self.request.user

        # get email
        email = user.email

        biosample_user_id = None

        # creating a user
        logger.debug("Creating user %s" % (form.username))

        try:
            biosample_user_id = User.create_user(
                user=form.username,
                password=password,
                confirmPwd=confirmPwd,
                email=email,
                full_name=full_name,
                organisation=affiliation
            )

            logger.info("user_id %s generated" % (biosample_user_id))

        except USIConnectionError as e:
            message = "Problem in creating user %s" % (form.username)
            self.deal_with_errors(message, e)

        return biosample_user_id

    def create_biosample_team(self, full_name, affiliation):
        """Create a biosample team

        Args:
            full_name (str): the user full name (Name + Surname)
            affiliation (str): the organization the user belongs to

        Returns:
            :py:class:`pyUSIrest.usi.Team`: a pyUSIrest Team instance
            :py:class:`biosample.models.ManagedTeam`: a model object
        """

        # creating a new team. First create an user object
        # create a new auth object
        logger.debug("Generate a token for 'USI_MANAGER'")

        # get an Auth instance from a helpers.method
        auth = get_manager_auth()
        admin = User(auth)

        description = "Team for %s" % (full_name)

        # the values I want to return
        team, managed_team = None, None

        # now create a team
        logger.debug("Creating team for %s" % (full_name))

        try:

            team = admin.create_team(
                description=description,
                centreName=affiliation)

            logger.info("Team %s generated" % (team.name))

            # register team in ManagedTeam table
            managed_team, created = ManagedTeam.objects.get_or_create(
                name=team.name)

            if created is True:
                logger.info("Created: %s" % (managed_team))

        except USIConnectionError as e:
            message = "Problem in creating team for %s" % (full_name)
            self.deal_with_errors(message, e)

        return team, managed_team

    def add_biosample_user_to_team(self, form, user_id, team, managed_team):

        # I need to generate a new token to see the new team
        logger.debug("Generate a new token for 'USI_MANAGER'")

        auth = get_manager_auth()
        admin = User(auth)

        # list all domain for manager
        logger.debug("Listing all domains for %s" % (config('USI_MANAGER')))
        logger.debug(", ".join(auth.get_domains()))

        # get the domain for the new team, and then the domain_id
        logger.debug("Getting domain info for team %s" % (team.name))
        domain = admin.get_domain_by_name(team.name)
        domain_id = domain.domainReference

        # the value I want to return
        account = None

        logger.debug(
            "Adding user %s to team %s" % (form.username, team.name))

        try:
            # user to team
            admin.add_user_to_team(user_id=user_id, domain_id=domain_id)

            # save objects in accounts table
            account = Account.objects.create(
                user=self.request.user,
                name=form.username,
                team=managed_team
            )

            logger.info("%s created" % (account))

            # add message
            messages.debug(
                request=self.request,
                message="User %s added to team %s" % (
                    form.username, team.name),
                extra_tags="alert alert-dismissible alert-light")

        except USIConnectionError as e:
            message = "Problem in adding user %s to team %s" % (
                form.username, team.name)
            self.deal_with_errors(message, e)

        return account

    # HINT: move to a celery task?
    def form_valid(self, form):
        """Create a new team in with biosample manager user, then create a new
        user and register it"""

        # get user model associated with this session
        user = self.request.user

        # get user info
        first_name = user.first_name
        last_name = user.last_name

        # set full name as
        full_name = " ".join([first_name, last_name])

        # Affiliation is the institution the user belong
        affiliation = user.person.affiliation.name

        # model generic errors from EBI API endpoints
        try:
            user_id = self.create_biosample_user(form, full_name, affiliation)

            if user_id is None:
                # return invalid form
                return self.form_invalid(form)

            # create a biosample team
            team, managed_team = self.create_biosample_team(
                full_name, affiliation)

            if team is None:
                # return invalid form
                return self.form_invalid(form)

            account = self.add_biosample_user_to_team(
                form, user_id, team, managed_team)

            if account is None:
                # return invalid form
                return self.form_invalid(form)

        except USIConnectionError as e:
            message = (
                "Problem with EBI-AAP endoints. Please contact IMAGE team")
            self.deal_with_errors(message, e)

            # return invalid form
            return self.form_invalid(form)

        # call to a inherited function (which does an HttpResponseRedirect
        # to success_url)
        return super(CreateUserView, self).form_valid(form)


class SubmitView(LoginRequiredMixin, TokenMixin, MyFormMixin, FormView):
    """Call a submission task. Check that a token exists and that it's valid"""

    form_class = SubmitForm
    template_name = 'biosample/submit.html'
    submission_id = None

    def get_success_url(self):
        return reverse('submissions:detail', kwargs={
            'pk': self.submission_id})

    def form_valid(self, form):
        submission_id = form.cleaned_data['submission_id']
        name = form.cleaned_data['name']
        password = form.cleaned_data['password']

        # get a submission object
        submission = get_object_or_404(
            Submission,
            pk=submission_id,
            owner=self.request.user)

        # track submission id in order to render page
        self.submission_id = submission_id

        # check if I can submit object (statuses)
        if not submission.can_submit():
            # return super method (which calls get_success_url)
            logger.error(
                "Can't submit submission %s: current status is %s" % (
                    submission, submission.get_status_display()))

            return HttpResponseRedirect(self.get_success_url())

        # create an auth object if credentials are provided
        if name and password:
            # call AuthMixin method and generate token. Check user/password
            auth = self.generate_token(form)

            if not auth:
                return self.form_invalid(form)

        # check token: if expired or near to expiring, return form
        elif 'token' in self.request.session:
            auth = get_auth(token=self.request.session['token'])

        else:
            logger.warning(
                "No valid token found. Redirect to tocken generation")

            messages.error(
                self.request,
                ("You haven't generated any token yet. Generate a new one"),
                extra_tags="alert alert-dismissible alert-danger")

            # redirect to this form
            return self.form_invalid(form)

        # check tocken expiration
        if (auth.is_expired() or
                auth.get_duration().seconds <= TOKEN_DURATION_THRESHOLD):
            logger.warning(
                "Token is expired or near to expire")

            messages.error(
                self.request,
                ("Your token is expired or near to expire. Generate a new "
                 "one"),
                extra_tags="alert alert-dismissible alert-danger")

            # redirect to this form
            return self.form_invalid(form)

        # start the submission with a valid token
        self.start_submission(auth, submission)

        return redirect(self.get_success_url())

    def start_submission(self, auth, submission):
        """Change submission status and submit data with a valid token"""

        logger.debug("Connecting to redis database")

        # here token is valid, so store it in redis database
        client = redis.StrictRedis(
            host=settings.REDIS_HOST,
            port=settings.REDIS_PORT,
            db=settings.REDIS_DB)

        key = "token:submission:{submission_id}:{user}".format(
            submission_id=self.submission_id,
            user=self.request.user)

        logger.debug("Writing token in redis")

        client.set(key, auth.token, ex=auth.get_duration().seconds)

        # Update submission status
        submission.status = WAITING
        submission.message = "Waiting for biosample submission"
        submission.save()

        # a valid submission start a task
        submit_task = SplitSubmissionTask()
        res = submit_task.delay(submission.id)
        logger.info(
            "Start submission process for %s with task %s" % (
                submission,
                res.task_id))


1
2			import os
3			import re
4			import json
5			import redis
6			import logging
7			import traceback
8
9			from decouple import AutoConfig
10
11			from django.conf import settings
12			from django.urls import reverse_lazy, reverse
13			from django.utils.crypto import get_random_string
14			from django.utils.http import is_safe_url
15			from django.shortcuts import redirect, get_object_or_404
16			from django.contrib.auth.mixins import LoginRequiredMixin
17			from django.views.generic import TemplateView
18			from django.views.generic.edit import FormView, CreateView, ModelFormMixin
19			from django.contrib import messages
20			from django.http import HttpResponseRedirect
21
22			from pyUSIrest.usi import User
23			from pyUSIrest.exceptions import USIConnectionError
24
25			from common.constants import WAITING
26			from common.helpers import send_mail_to_admins
27			from uid.models import Submission
28
29			from .forms import (
30			GenerateTokenForm, RegisterUserForm, CreateUserForm, SubmitForm)
31			from .models import Account, ManagedTeam
32			from .helpers import get_auth, get_manager_auth
33			from .tasks import SplitSubmissionTask
34
35
36			# Get an instance of a logger
37			logger = logging.getLogger(__name__)
38
39			# while near
40			TOKEN_DURATION_THRESHOLD = 3600*4
41
42			# define a decouple config object
43			settings_dir = os.path.join(settings.BASE_DIR, 'image')
44			config = AutoConfig(search_path=settings_dir)
45
46
47			# In programming, a mixin is a class that provides functionality to be
48			# inherited, but isn’t meant for instantiation on its own. In programming
49			# languages with multiple inheritance, mixins can be used to add enhanced
50			# functionality and behavior to classes.
51			class AccountMixin(object):
52			"""A generic mixin associated with biosample.models. You need to costomize
53			account_found and account_not_found in to do a custom redirect in case
54			a manager account is found or not"""
55
56			def account_not_found(self, request, args, *kwargs):
57			return super().dispatch(request, args, *kwargs)
58
59			def account_found(self, request, args, *kwargs):
60			return super().dispatch(request, args, *kwargs)
61
62			def dispatch(self, request, args, *kwargs):
63			# get user from request and user model. This could be also Anonymous
64			# user:however with metod decorator a login is required before dispatch
65			# method is called
66			user = self.request.user
67
68			if hasattr(user, "biosample_account"):
69			return self.account_found(request, args, *kwargs)
70
71			else:
72			return self.account_not_found(request, args, *kwargs)
73
74
75			class TokenMixin(AccountMixin):
76			"""Get common stuff for Token visualization. Redirect to AAP registration
77			if no valid AAP credentials are found for request.user"""
78
79			def get_initial(self):
80			"""
81			Returns the initial data to use for forms on this view.
82			"""
83
84			initial = super(TokenMixin, self).get_initial()
85			initial['name'] = self.request.user.biosample_account.name
86
87			return initial
88
89			# override AccountMixin method
90			def account_not_found(self, request, args, *kwargs):
91			"""If a user has not an account, redirect to activation complete"""
92
93			logger.warning("Error for user:%s: not managed" % self.request.user)
94
95			messages.warning(
96			request=self.request,
97			message='You need to register a valid AAP profile',
98			extra_tags="alert alert-dismissible alert-warning")
99
100			return redirect('accounts:registration_activation_complete')
101
102
103			class RegisterMixin(AccountMixin):
104			"""If a biosample account is already registered, returns to dashboard"""
105
106			# override AccountMixin method
107			def account_found(self, request, args, *kwargs):
108			"""If a user has been registered, redirect to dashboard"""
109
110			logger.warning(
111			"Error for user:%s: Already registered" % self.request.user)
112
113			messages.warning(
114			request=self.request,
115			message='Your AAP profile is already registered',
116			extra_tags="alert alert-dismissible alert-warning")
117
118			return redirect('uid:dashboard')
119
120
121			class MyFormMixin(object):
122			"""Common stuff for token generation"""
123
124			success_url_message = "Please set this variable"
125			success_url = reverse_lazy("uid:dashboard")
126
127			# add the request to the kwargs
128			# https://chriskief.com/2012/12/18/django-modelform-formview-and-the-request-object/
129			# this is needed to display messages (django.contronb) on pages
130			def get_form_kwargs(self):
131			kwargs = super(MyFormMixin, self).get_form_kwargs()
132			kwargs['request'] = self.request
133			return kwargs
134
135			def get_success_url(self):
136			"""Override default function"""
137
138			messages.success(
139			request=self.request,
140			message=self.success_url_message,
141			extra_tags="alert alert-dismissible alert-success")
142
143			return super(MyFormMixin, self).get_success_url()
144
145			def form_invalid(self, form):
146			messages.error(
147			self.request,
148			message="Please correct the errors below",
149			extra_tags="alert alert-dismissible alert-danger")
150
151			return super(MyFormMixin, self).form_invalid(form)
152
153			def generate_token(self, form):
154			"""Generate token from form instance"""
155
156			# operate on form data
157			name = form.cleaned_data['name']
158			password = form.cleaned_data['password']
159
160			try:
161			auth = get_auth(user=name, password=password)
162
163			except USIConnectionError as e:
164			# logger exception. With repr() the exception name is rendered
165			logger.error(repr(e))
166
167			# try to detect the error message
168			pattern = re.compile(r"(\{.*\})")
169			match = re.search(pattern, str(e))
170
171			if match is None:
172			# parse error message
173			messages.error(
174			self.request,
175			"Unable to generate token: %s" % str(e),
176			extra_tags="alert alert-dismissible alert-danger")
177
178			else:
179			data = json.loads(match.groups()[0])
180			messages.error(
181			self.request,
182			"Unable to generate token: %s" % data['message'],
183			extra_tags="alert alert-dismissible alert-danger")
184
185			# cant't return form_invalid here, since i need to process auth
186			return None
187
188			else:
189			logger.debug("Token generated for user:%s" % name)
190
191			self.request.session['token'] = auth.token
192
193			# return an auth object
194			return auth
195
196
197			class GenerateTokenView(LoginRequiredMixin, TokenMixin, MyFormMixin, FormView):
198			"""Generate AAP token. If user is not registered, redirect to accounts
199			registration_activation_complete through TokenMixin. If yes generate
200			token through MyFormMixin"""
201
202			template_name = 'biosample/generate_token.html'
203			form_class = GenerateTokenForm
204			success_url_message = 'Token generated!'
205
206			def dispatch(self, request, args, *kwargs):
207			# try to read next link
208			next_url = request.GET.get('next', None)
209
210			# redirect to next url. is_safe_url: is a safe redirection
211			# (i.e. it doesn't point to a different host and uses a safe scheme).
212			if next_url and is_safe_url(next_url, allowed_hosts=None):
213			logger.debug("Got %s as next_url" % next_url)
214			self.success_url = next_url
215
216			return super(
217			GenerateTokenView, self).dispatch(request, args, *kwargs)
218
219			def form_valid(self, form):
220			# This method is called when valid form data has been POSTed.
221			# It should return an HttpResponse.
222
223			# call MyFormMixin method and generate token. Check user/password
224			if not self.generate_token(form):
225			return self.form_invalid(form)
226
227			return redirect(self.get_success_url())
228
229
230			class TokenView(LoginRequiredMixin, TokenMixin, TemplateView):
231			"""Visualize token details"""
232
233			template_name = 'biosample/token.html'
234
235			def get_context_data(self, **kwargs):
236			# Call the base implementation first to get a context
237			context = super(TokenView, self).get_context_data(**kwargs)
238
239			# get user and team object
240			context['name'] = self.request.user.biosample_account.name
241			context['team'] = self.request.user.biosample_account.team
242
243			try:
244			# add content to context
245			auth = get_auth(token=self.request.session['token'])
246
247			if auth.is_expired():
248			messages.error(
249			self.request,
250			"Your token is expired",
251			extra_tags="alert alert-dismissible alert-danger")
252
253			context["auth"] = auth
254
255			except KeyError as e:
256			logger.error(
257			"No valid token found for %s: %s" % (
258			self.request.user,
259			e))
260
261			messages.error(
262			self.request,
263			"You haven't generated any token yet",
264			extra_tags="alert alert-dismissible alert-danger")
265
266			return context
267
268
269			class RegisterUserView(LoginRequiredMixin, RegisterMixin, MyFormMixin,
270			CreateView):
271			"""Register an already existent AAP account"""
272
273			template_name = 'biosample/register_user.html'
274			form_class = RegisterUserForm
275			success_url_message = 'Account registered'
276
277			def form_valid(self, form):
278			# This method is called when valid form data has been POSTed.
279			# It should return an HttpResponse.
280			team = form.cleaned_data['team']
281
282			# call AuthMixin method and generate token. Check user/password
283			auth = self.generate_token(form)
284
285			if not auth:
286			return self.form_invalid(form)
287
288			if team.name not in auth.get_domains():
289			messages.error(
290			self.request,
291			"You don't belong to team: %s" % team,
292			extra_tags="alert alert-dismissible alert-danger")
293
294			# return invalid form
295			return self.form_invalid(form)
296
297			# add a user to object (comes from section not from form)
298			self.object = form.save(commit=False)
299			self.object.user = self.request.user
300			self.object.save()
301
302			# call to a specific function (which does an HttpResponseRedirect
303			# to success_url)
304			return super(ModelFormMixin, self).form_valid(form)
305
306
307			class CreateUserView(LoginRequiredMixin, RegisterMixin, MyFormMixin, FormView):
308			"""Create a new AAP account"""
309
310			template_name = 'biosample/create_user.html'
311			form_class = CreateUserForm
312			success_url_message = "AAP Account created"
313
314			def deal_with_errors(self, error_message, exception):
315			"""Add messages to view for encountered errors
316
317			Args:
318			error_message (str): An informative text
319			exception (Exception): an exception instance
320			"""
321
322			exception_message = "Message was: %s" % (exception)
323
324			logger.error(error_message)
325			logger.error(exception_message)
326
327			messages.error(
328			self.request,
329			message=error_message,
330			extra_tags="alert alert-dismissible alert-danger")
331
332			messages.error(
333			self.request,
334			message=exception_message,
335			extra_tags="alert alert-dismissible alert-danger")
336
337			# get exception info
338			einfo = traceback.format_exc()
339
340			# send a mail to admin
341			send_mail_to_admins(error_message, einfo)
342
343			def get_form_kwargs(self):
344			"""Override get_form_kwargs"""
345
346			kwargs = super(CreateUserView, self).get_form_kwargs()
347
348			# create a new biosample user
349			username = "image-%s" % (get_random_string(length=8))
350
351			# add username to instance
352			kwargs['username'] = username
353
354			return kwargs
355
356			# create an user or throw an exception
357			def create_biosample_user(self, form, full_name, affiliation):
358			"""Create a biosample user
359
360			Args:
361			form (:py:class:`CreateUserForm`) an instantiated form object
362			full_name (str): the user full name (Name + Surname)
363			affiliation (str): the organization the user belongs to
364
365			Returns:
366			str: a biosamples user id
367			"""
368
369			password = form.cleaned_data['password1']
370			confirmPwd = form.cleaned_data['password2']
371
372			# get user model associated with this session
373			user = self.request.user
374
375			# get email
376			email = user.email
377
378			biosample_user_id = None
379
380			# creating a user
381			logger.debug("Creating user %s" % (form.username))
382
383			try:
384			biosample_user_id = User.create_user(
385			user=form.username,
386			password=password,
387			confirmPwd=confirmPwd,
388			email=email,
389			full_name=full_name,
390			organisation=affiliation
391			)
392
393			logger.info("user_id %s generated" % (biosample_user_id))
394
395			except USIConnectionError as e:
396			message = "Problem in creating user %s" % (form.username)
397			self.deal_with_errors(message, e)
398
399			return biosample_user_id
400
401			def create_biosample_team(self, full_name, affiliation):
402			"""Create a biosample team
403
404			Args:
405			full_name (str): the user full name (Name + Surname)
406			affiliation (str): the organization the user belongs to
407
408			Returns:
409			:py:class:`pyUSIrest.usi.Team`: a pyUSIrest Team instance
410			:py:class:`biosample.models.ManagedTeam`: a model object
411			"""
412
413			# creating a new team. First create an user object
414			# create a new auth object
415			logger.debug("Generate a token for 'USI_MANAGER'")
416
417			# get an Auth instance from a helpers.method
418			auth = get_manager_auth()
419			admin = User(auth)
420
421			description = "Team for %s" % (full_name)
422
423			# the values I want to return
424			team, managed_team = None, None
425
426			# now create a team
427			logger.debug("Creating team for %s" % (full_name))
428
429			try:
430
431			team = admin.create_team(
432			description=description,
433			centreName=affiliation)
434
435			logger.info("Team %s generated" % (team.name))
436
437			# register team in ManagedTeam table
438			managed_team, created = ManagedTeam.objects.get_or_create(
439			name=team.name)
440
441			if created is True:
442			logger.info("Created: %s" % (managed_team))
443
444			except USIConnectionError as e:
445			message = "Problem in creating team for %s" % (full_name)
446			self.deal_with_errors(message, e)
447
448			return team, managed_team
449
450			def add_biosample_user_to_team(self, form, user_id, team, managed_team):
451
452			# I need to generate a new token to see the new team
453			logger.debug("Generate a new token for 'USI_MANAGER'")
454
455			auth = get_manager_auth()
456			admin = User(auth)
457
458			# list all domain for manager
459			logger.debug("Listing all domains for %s" % (config('USI_MANAGER')))
460			logger.debug(", ".join(auth.get_domains()))
461
462			# get the domain for the new team, and then the domain_id
463			logger.debug("Getting domain info for team %s" % (team.name))
464			domain = admin.get_domain_by_name(team.name)
465			domain_id = domain.domainReference
466
467			# the value I want to return
468			account = None
469
470			logger.debug(
471			"Adding user %s to team %s" % (form.username, team.name))
472
473			try:
474			# user to team
475			admin.add_user_to_team(user_id=user_id, domain_id=domain_id)
476
477			# save objects in accounts table
478			account = Account.objects.create(
479			user=self.request.user,
480			name=form.username,
481			team=managed_team
482			)
483
484			logger.info("%s created" % (account))
485
486			# add message
487			messages.debug(
488			request=self.request,
489			message="User %s added to team %s" % (
490			form.username, team.name),
491			extra_tags="alert alert-dismissible alert-light")
492
493			except USIConnectionError as e:
494			message = "Problem in adding user %s to team %s" % (
495			form.username, team.name)
496			self.deal_with_errors(message, e)
497
498			return account
499
500			# HINT: move to a celery task?
501			def form_valid(self, form):
502			"""Create a new team in with biosample manager user, then create a new
503			user and register it"""
504
505			# get user model associated with this session
506			user = self.request.user
507
508			# get user info
509			first_name = user.first_name
510			last_name = user.last_name
511
512			# set full name as
513			full_name = " ".join([first_name, last_name])
514
515			# Affiliation is the institution the user belong
516			affiliation = user.person.affiliation.name
517
518			# model generic errors from EBI API endpoints
519			try:
520			user_id = self.create_biosample_user(form, full_name, affiliation)
521
522			if user_id is None:
523			# return invalid form
524			return self.form_invalid(form)
525
526			# create a biosample team
527			team, managed_team = self.create_biosample_team(
528			full_name, affiliation)
529
530			if team is None:
531			# return invalid form
532			return self.form_invalid(form)
533
534			account = self.add_biosample_user_to_team(
535			form, user_id, team, managed_team)
536
537			if account is None:
538			# return invalid form
539			return self.form_invalid(form)
540
541			except USIConnectionError as e:
542			message = (
543			"Problem with EBI-AAP endoints. Please contact IMAGE team")
544			self.deal_with_errors(message, e)
545
546			# return invalid form
547			return self.form_invalid(form)
548
549			# call to a inherited function (which does an HttpResponseRedirect
550			# to success_url)
551			return super(CreateUserView, self).form_valid(form)
552
553
554			class SubmitView(LoginRequiredMixin, TokenMixin, MyFormMixin, FormView):
555			"""Call a submission task. Check that a token exists and that it's valid"""
556
557			form_class = SubmitForm
558			template_name = 'biosample/submit.html'
559			submission_id = None
560
561			def get_success_url(self):
562			return reverse('submissions:detail', kwargs={
563			'pk': self.submission_id})
564
565			def form_valid(self, form):
566			submission_id = form.cleaned_data['submission_id']
567			name = form.cleaned_data['name']
568			password = form.cleaned_data['password']
569
570			# get a submission object
571			submission = get_object_or_404(
572			Submission,
573			pk=submission_id,
574			owner=self.request.user)
575
576			# track submission id in order to render page
577			self.submission_id = submission_id
578
579			# check if I can submit object (statuses)
580			if not submission.can_submit():
581			# return super method (which calls get_success_url)
582			logger.error(
583			"Can't submit submission %s: current status is %s" % (
584			submission, submission.get_status_display()))
585
586			return HttpResponseRedirect(self.get_success_url())
587
588			# create an auth object if credentials are provided
589			if name and password:
590			# call AuthMixin method and generate token. Check user/password
591			auth = self.generate_token(form)
592
593			if not auth:
594			return self.form_invalid(form)
595
596			# check token: if expired or near to expiring, return form
597			elif 'token' in self.request.session:
598			auth = get_auth(token=self.request.session['token'])
599
600			else:
601			logger.warning(
602			"No valid token found. Redirect to tocken generation")
603
604			messages.error(
605			self.request,
606			("You haven't generated any token yet. Generate a new one"),
607			extra_tags="alert alert-dismissible alert-danger")
608
609			# redirect to this form
610			return self.form_invalid(form)
611
612			# check tocken expiration
613			if (auth.is_expired() or
614			auth.get_duration().seconds <= TOKEN_DURATION_THRESHOLD):
615			logger.warning(
616			"Token is expired or near to expire")
617
618			messages.error(
619			self.request,
620			("Your token is expired or near to expire. Generate a new "
621			"one"),
622			extra_tags="alert alert-dismissible alert-danger")
623
624			# redirect to this form
625			return self.form_invalid(form)
626
627			# start the submission with a valid token
628			self.start_submission(auth, submission)
629
630			return redirect(self.get_success_url())
631
632			def start_submission(self, auth, submission):
633			"""Change submission status and submit data with a valid token"""
634
635			logger.debug("Connecting to redis database")
636
637			# here token is valid, so store it in redis database
638			client = redis.StrictRedis(
639			host=settings.REDIS_HOST,
640			port=settings.REDIS_PORT,
641			db=settings.REDIS_DB)
642
643			key = "token:submission:{submission_id}:{user}".format(
644			submission_id=self.submission_id,
645			user=self.request.user)
646
647			logger.debug("Writing token in redis")
648
649			client.set(key, auth.token, ex=auth.get_duration().seconds)
650
651			# Update submission status
652			submission.status = WAITING
653			submission.message = "Waiting for biosample submission"
654			submission.save()
655
656			# a valid submission start a task
657			submit_task = SplitSubmissionTask()
658			res = submit_task.delay(submission.id)
659			logger.info(
660			"Start submission process for %s with task %s" % (
661			submission,
662			res.task_id))
663

cnr-ibba / IMAGE-InjectTool

biosample.views.CreateUserView.form_valid() B last analyzed 2023-06-12 09:55 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

biosample.views.CreateUserView.form_valid() B
last analyzed 2023-06-12 09:55 UTC