Python django.contrib.auth.models.User 模块，objects() 实例源码

def test_logged_in_user_can_edit_self(self):
        """GET /user/{pk} A user should be able to edit their own details"""

        self.c.login(username="joe", password="password")
        url = reverse("user-detail", args=[self.normal_user.pk])

        data = {
            "username": "joe",
            "first_name": "Joe", 
            "last_name": "Soap",
        }

        response = self.c.put(url, data, format="json")

        assert response.status_code == 200, \
            "Expect 200 OK. got: {}: {}" . format(response.status_code, response.content)

        joe = User.objects.get(username="joe")

        assert joe.first_name == "Joe", \
            "Expect user's first_name to be Joe. Got: {}" . format(joe.first_name)

def homepage(request):
    """
    Homepage view for when the URL does not specify a specific semester.
    Looks at session data to see the user's last visited semester.
    If no data is given, the homepage defaults to the semester
    given by DEFAULT_SEMESTER_PK
    """
    if request.user.is_authenticated:
        return redirect(to=request.user.options.get_absolute_url())
    elif request.session.get('homepage', ''):
        # If the student has visited a student page before, redirect
        return redirect(django_reverse(
            'semesterpage-studyprogram',
            args=(request.session.get('homepage'),)
        ))
    else:
        semester_pk = request.session.get('semester_pk', DEFAULT_SEMESTER_PK)
        semester = Semester.objects.get(pk=semester_pk)
        return redirect(to=semester.get_absolute_url())

def new_course_url(request, course_pk: str) -> HttpResponse:
    """
    A user has specified a URL for a course which previously had none.
    This should be saved to the Course model object before redirecting
    to the course homepage.
    """
    homepage_url = request.GET.get('homepage_url', '')

    # Need to prevent relative links
    if not homepage_url[:4].lower() == 'http':
        homepage_url = 'http://' + homepage_url

    course = Course.objects.get(pk=int(course_pk))
    course.homepage = homepage_url.strip()
    course.save(update_fields=['homepage'])
    return redirect(course.homepage)

def get_queryset(self):
        """
        Returns a queryset used for autocompletion, restricted based
        on the user input.
        """
        # Only let logged in users access all the course information
        if not self.request.user.is_authenticated():
            return Course.objects.none()

        qs = Course.objects.all()

        # If the user has started entering input, start restricting
        # the choices available for autocompletion.
        if self.q:
            course_code = Q(course_code__istartswith=self.q)
            full_name = Q(full_name__istartswith=self.q)
            display_name = Q(display_name__startswith=self.q)
            qs = qs.filter(course_code | full_name | display_name)

        return qs

def get(self, request, *args, **kwargs):
        # users = (User.objects
        #          .filter(is_active = 1)
        #          .values('id', 'contributor__full_name')
        #          .annotate(Count("id"), poems_count = Count('poem__id'))
        #          .order_by('-poems_count'))
        users = User.objects.raw('''
                SELECT
                    users.id, uc.full_name as contributor__full_name, COUNT(poem.id) as poems_count
                FROM
                    auth_user as users
                JOIN user_contributor as uc ON uc.user_id=users.id
                JOIN poetry_poem as poem ON poem.added_user_id=users.id AND poem.is_shown=1
                WHERE users.is_active=1
                GROUP BY users.id
                ORDER BY poems_count DESC
            ''')
        return render(request, self.template_name, {
            'users': users
        })

def __init__(
            self, data=None, instance=None, request=None, *args, **kwargs):
        if data is None:
            data = {}
        else:
            data = data.copy()
        # If this is a new IPRange, fill in the 'user' and 'subnet' fields
        # automatically, if necessary.
        if instance is None:
            start_ip = data.get('start_ip')
            subnet = data.get('subnet')
            if subnet is None and start_ip is not None:
                subnet = Subnet.objects.get_best_subnet_for_ip(start_ip)
                if subnet is not None:
                    data['subnet'] = subnet.id
            if request is not None:
                data['user'] = request.user.username
        elif instance.user and 'user' not in data:
            data['user'] = instance.user.username
        super().__init__(
            data=data, instance=instance, *args, **kwargs)

def set_choices(self,user):
        """ Sets the mitspieler choices to all other non-admin users"""
        choices = []
        initial = []
        users = User.objects\
            .annotate(lastpart=Max('teller__storyparts__creation'))\
            .filter(is_superuser=False, is_active=True)\
            .exclude(id=user.id).all()
        for u in users:
            if u.lastpart:
                last_storypart_date = u.lastpart.strftime("%Y-%m-%d")
            else:
                last_storypart_date = "never"
            assert u != user and not u.is_superuser and u.is_active
            initial.append(u.pk)
            choices.append((u.pk, "%s   (last active: %s)" % (str(u), last_storypart_date)))
        self.fields['mitspieler'].choices = choices
        self.fields['mitspieler'].initial = []

def create_new_story(request):
    form = NewStoryForm(request.POST)
    form.set_choices(request.user)
    if form.is_valid():
        users = [ User.objects.get(pk=uid) for uid in form.cleaned_data['mitspieler'] ]
        shuffle(users)
        s = Story.create_new_story(
            title = form.cleaned_data['title'],
            rules = form.cleaned_data['rules'],
            startUser = request.user,
            first_sentence = form.cleaned_data['firstSentence'],
            participating_users = users
            )
        messages.success(request, u"Geschichte „%s“ gestartet" % s.title)
        return redirect('overview')
    else:
        context = {
            'form': form
        }
        return render(request, 'umklapp/start_story.html', context)

def user_profile(request):
    form = UserUpdateForm(instance=request.user)
    context = {
        'form': form,
        'sentences_written': StoryPart.objects.filter(teller__user=request.user).count(),
        'participated_in': Teller.objects.filter(user=request.user).count(),
        'stories_started': Story.objects.filter(started_by=request.user).count(),
        'funny_parts': StoryPart.objects.select_related('teller__user') \
                .select_related('teller__corresponding_story') \
                .filter(teller__user=request.user) \
                .annotate(Count('upvotes')) \
                .filter(upvotes__count__gt=0) \
                .order_by('-upvotes__count','-teller__corresponding_story__id') \
                [:10]
    }
    return render(request, 'umklapp/profile.html', context)

def handle(self, *args, **options):
        cutoff = timezone.now() - timedelta(days=31)

        # Old accounts, never logged in
        q = User.objects
        q = q.filter(date_joined__lt=cutoff, last_login=None)
        n1, _ = q.delete()

        # Not logged in for 1 month, 0 checks
        q = User.objects
        q = q.annotate(n_checks=Count("check"))
        q = q.filter(last_login__lt=cutoff, n_checks=0)
        n2, _ = q.delete()

        return "Done! Pruned %d user accounts." % (n1 + n2)

def unseed_db():
    """
    Deletes all seed data from the database
    """
    fake_program_ids = (
        Program.objects
        .filter(description__startswith=FAKE_PROGRAM_DESC_PREFIX)
        .values_list('id', flat=True)
    )
    fake_user_ids = (
        User.objects
        .filter(username__startswith=FAKE_USER_USERNAME_PREFIX)
        .values_list('id', flat=True)
    )
    fake_tier_ids = (
        TierProgram.objects
        .filter(program__id__in=fake_program_ids)
        .values_list('tier__id', flat=True)
    )
    fake_final_grade_ids = (
        FinalGrade.objects
        .filter(course_run__course__program__id__in=fake_program_ids)
        .values_list('id', flat=True)
    )
    financial_aid_ids = (
        FinancialAid.objects
        .filter(Q(user_id__in=fake_user_ids) | Q(tier_program__program__id__in=fake_program_ids))
        .values_list('id', flat=True)
    )
    fin_aid_audit_models = [FinancialAidAudit, FinancialAidEmailAudit]
    with mute_signals(post_delete):
        with remove_delete_protection(*fin_aid_audit_models):
            for audit_model in fin_aid_audit_models:
                audit_model.objects.filter(financial_aid__id__in=financial_aid_ids).delete()
        for model_cls in [CachedEnrollment, CachedCertificate, CachedCurrentGrade]:
            model_cls.objects.filter(course_run__course__program__id__in=fake_program_ids).delete()
        Tier.objects.filter(id__in=fake_tier_ids).delete()
        FinalGrade.objects.filter(id__in=fake_final_grade_ids).delete()
        Program.objects.filter(id__in=fake_program_ids).delete()
        User.objects.filter(id__in=fake_user_ids).delete()

def get_queryset(self):
        return Place.objects.all()

    ########Random places

def random(self, request, *args, **kwargs):
        count = Place.objects.count()
        random = randint(0, count-10 )
        places = Place.objects(Q(types__contains="restaurant") | Q(types__contains= "bar")| Q(types__contains= "food")|
                               Q(types__contains= "bakery") | Q(types__contains= "cafe")| Q(types__contains= "casino") |
                               Q(types__contains= "convenience_store") | Q(types__contains= "meal_delivery") |
                               Q(types__contains= "make_takeaway") | Q(types__contains= "nightclub") |
                               Q(types__contains= "shopping_mall")).limit(10).skip(random)
        serializer = PlaceSerializer(places, many=True)
        return Response(serializer.data)


############################
######RatedPlaceViewSet#####
############################

def get_queryset(self):
        return RatedPlace.objects.all()

def get_queryset(self):
        return Category.objects.all()


############################
########UserViewSet#########
############################

def get_queryset(self):
        return User.objects.all()

def categories(self, request, *args, **kwargs):

        snippet = self.get_object()
        userId = kwargs.get("id")
        user = User.objects(id=userId).get()
        userTypes = user.types
        listCategories = []

        for type in userTypes:
            type = Category.objects(id=type).get()
            listCategories.append(type)

        serializer = CategorySerializer(listCategories, many=True)

        return Response(serializer.data)

def autocomplete(request, queryset_name=None):
    term = request.GET.get('term', '')

    USER_QUERY = {
        'users': Q(),
        'internal-users': Q(profile__is_internal=True),
        'pki-users':
            Q(profile__is_internal=True) | Q(profile__is_omniscient_member=True),
    }

    users = (User.objects
        .filter(USER_QUERY[queryset_name], is_active=True)
        .select_related('profile')
        .order_by('first_name', 'last_name', 'email')
    )

    if term:
        q = Q()
        for bit in term.split():
            q &= (
                Q(first_name__icontains=bit) |
                Q(last_name__icontains=bit) |
                Q(email__icontains=bit) |
                Q(profile__title__icontains=bit)
            )
        users = users.filter(q)

    results = [{'id': u.id, 'text': '{} [{}]'.format(u, u.email)} for u in users]
    return JsonResponse({'results': results})

def studentpage(request, homepage):
    # NB! In the following view function, user and request.user is not necessarily
    # the same user. This should become more clear in the next refactoring
    try:
        # The homepage is given by the (Feide) username
        user = User.objects\
                .select_related('options', 'contributor')\
                .prefetch_related('options__self_chosen_courses__links')\
                .get(username=homepage)

    except User.DoesNotExist:
        raise Http404(_('Fant ingen studieprogram eller brukerside med navnet "%s"') % homepage)

    if request.user.is_authenticated \
            and isinstance(request.user, DataportenUser) \
            and request.user.username == homepage:
        # We ensure normalization between dataporten and the database,
        # since we have an authenticated dataporten user which tries to acces
        # his/her own homepage.
        reconcile_dataporten_data(request.user)
        user.refresh_from_db()

    # Save homepage in session for automatic redirect on next visit
    request.session['homepage'] = homepage

    return render(request, 'semesterpage/userpage-courses.html', {
           'semester': user.options,
           'courses': user.options.courses,
           'study_programs': StudyProgram.objects.filter(published=True),
           'calendar_name': get_calendar_name(request),
           'user': request.user,
           'header_text': f' / {user.username}',
           'student_page': True,
       }
    )

def get(self, request, *args, **kwargs):
        contributor = Contributor.objects.filter(user_id=kwargs.get('pk'), user__is_active=1).get()

        return render(request, self.template_name, {
            'contributor': contributor
        })

def get(self, request, *args, **kwargs):
        user_id = kwargs.get('id')
        activation_code = kwargs.get('hash')
        message = ''
        css_class = 'alert-success'

        contributor = Contributor.objects.filter(user_id=user_id).get()

        if contributor.activation_code != '' and contributor.user.is_active == 0:
            if contributor.activation_code != activation_code:
                message = '???????????? ??? ????? ????'
                css_class = 'alert-warning'
            else:
                contributor.activation_code = ''
                contributor.user.is_active = 1
                contributor.save()
                contributor.user.save()
                message = '????????? ????? ????. ??? ????????'
        else:
            message = '??? ?????? ??????????? ?????????'
            css_class = 'alert-info'

        return render(request, self.template_name, {
            'message': message,
            'css_class': css_class
        })

def get(self, request, *args, **kwargs):
        contributor = Contributor.objects.filter(user_id=kwargs.get('pk'), user__is_active=1).get()

        return render(request, self.template_name, {
            'contributor': contributor
        })

def get_object(self, queryset=None):
        if int(self.request.user.id) != int(self.kwargs['pk']):
            raise Http404("??? ??? ?? ????????????? ????? ??????")

        obj = Contributor.objects.get(user_id=self.request.user)

        return obj

def index(request):
    treasures = Treasure.objects.all()
    form = TreasureForm()
    return render(request, 'index.html', {'treasures': treasures, 'form':form})

def detail(request, treasure_id):
    treasure = Treasure.objects.get(id=treasure_id)
    #hashtags = HashTag.objects.filter(treasure=treasure_id)
    return render(request, 'detail.html',
                  {'treasure': treasure})

def profile(request, username):
    user = get_object_or_404(User.objects, username=username)
    treasures = Treasure.objects.filter(user=user)
    return render(request, 'profile.html', {'username':username,'treasures': treasures})

def like_treasure(request):
    treasure_id = request.POST.get('treasure_id', None)

    likes = 0
    if (treasure_id):
        treasure = Treasure.objects.get(id=int(treasure_id))
        if treasure is not None:
            likes = treasure.likes + 1
            treasure.likes = likes
            treasure.save()

    return HttpResponse(likes)

def tearDown(self):
        for user in User.objects.all():
            user.delete()

def tearDown(self):
        for user in User.objects.all():
            user.delete()

def continue_story(request, story_id):
    s = get_object_or_404(Story.objects, id=story_id)

    if s.is_finished:
        return HttpResponseBadRequest("Story already finished")

    if not s.participates_in(request.user):
        raise PermissionDenied

    t = get_object_or_404(s.tellers, user=request.user)

    if s.whose_turn != t.position:
        return HttpResponseBadRequest("Not your turn")

    finish = 'finish' in request.POST
    form = ExtendStoryForm(request.POST)
    if form.is_valid():
        if 'finish' in form.data:
            if form.cleaned_data['nextSentence']:
                s.continue_story(form.cleaned_data['nextSentence'])
            s.finish()
            return redirect('show_story', story_id=s.id)
        else:
            s.continue_story(form.cleaned_data['nextSentence'])
            messages.success(request, u"Geschichte „%s“ weitergeführt" % s.title)
            return redirect('overview')
    else:
        context = {
            'story': s,
            'form': form
        }
        return render(request, 'umklapp/extend_story.html', context)

def skip_always(request, story_id):
    s = get_object_or_404(Story.objects, id=story_id)
    u = request.user

    if s.is_finished:
        return HttpResponseBadRequest("Story already finished")
    if not s.participates_in(u):
        raise PermissionDenied

    try:
        s.set_always_skip(u)
        messages.success(request, u"Du wirst nun automatisch übersprungen bei „%s“." % s.title)
    except NotEnoughActivePlayers as e:
        messages.success(request, u"Zuwenig aktive Spieler, als dass du überspringen kannst bei „%s“." % s.title)
    return redirect('overview')

def unskip_always(request, story_id):
    s = get_object_or_404(Story.objects, id=story_id)
    u = request.user

    if not s.participates_in(u):
        raise PermissionDenied
    if s.is_finished:
        return HttpResponseBadRequest("Story already finished")

    messages.success(request, u"Du schreibst wieder aktiv mit bei „%s“." % s.title)
    s.unset_always_skip(u)
    return redirect('overview')

def show_story(request, story_id):
    s = get_object_or_404(Story.objects, id=story_id)

    if s.is_finished:
        if not s.participates_in(request.user) and not s.is_public:
            raise PermissionDenied

        anonym = not s.participates_in(request.user)

        if request.user.is_authenticated() and not request.user in s.read_by.all():
            s.read_by.add(request.user)

        context = {
            'story': s,
            'anonymized' : anonym,
            'has_upvoted' : s.has_upvoted(request.user),
            'upvote_count' : s.upvote_count(),
        }
        return render(request, 'umklapp/show_story.html', context)
    else:
        # unfinished business
        assert not s.is_finished

        if not s.participates_in(request.user):
            raise PermissionDenied

        form = None
        t = get_object_or_404(s.tellers, user=request.user)

        if s.whose_turn == t.position:
            # only show form if its the user's turn
            form = ExtendStoryForm()

        context = {
            'story': s,
            'form': form,
            'always_skip' : s.does_always_skip(request.user),
        }
        return render(request, 'umklapp/extend_story.html', context)

def update_profile(request):
    form = UserUpdateForm(request.POST, instance=request.user)
    if form.is_valid():
        form.save()
        messages.success(request, u"Daten gespeichert")
        return redirect('user_profile')
    else:
        context = {
            'form': form,
            'sentences_written': StoryPart.objects.filter(teller__user=request.user).count(),
            'participated_in': Teller.objects.filter(user=request.user).count(),
            'stories_started': Story.objects.filter(started_by=request.user).count(),
        }
        return render(request, 'umklapp/profile.html', context)

def upvote_story(request, story_id):
    s = get_object_or_404(Story.objects, id=story_id)

    if not s.is_finished:
        return HttpResponseBadRequest("Story not finished yet")

    s.upvote_story(request.user)

    return redirect('show_story', story_id=s.id)

def downvote_story(request, story_id):
    s = get_object_or_404(Story.objects, id=story_id)

    if not s.is_finished:
        return HttpResponseBadRequest("Story not finished yet")

    s.downvote_story(request.user)

    return redirect('show_story', story_id=s.id)

def downvote_storypart(request, storypart_id):
    s = get_object_or_404(StoryPart.objects, id=storypart_id)

    s.downvote(request.user)

    return redirect('show_story', story_id=s.teller.corresponding_story.id)

def publish_story(request, story_id):
    s = get_object_or_404(Story.objects, id=story_id)

    if not s.is_finished:
        return HttpResponseBadRequest("Story not finished yet")
    if not s.started_by == request.user:
        raise PermissionDenied

    s.public(True)

    return redirect('show_story', story_id=s.id)

def unpublish_story(request, story_id):
    s = get_object_or_404(Story.objects, id=story_id)

    if not s.is_finished:
        return HttpResponseBadRequest("Story not finished yet")
    if not s.started_by == request.user:
        raise PermissionDenied

    s.public(False)

    return redirect('show_story', story_id=s.id)

def running_stories(request):
    all_running_stories = Story.objects \
            .filter(is_finished = False) \
            .order_by('title', 'id') \
            .annotate(parts_count = Count('tellers__storyparts',distinct=True)) \
            .annotate(contrib_count = Count('tellers__storyparts__teller', distinct=True)) \
            .annotate(active_count = Count('tellers', distinct=True)) \
            .select_related('started_by') \
            .prefetch_related('always_skip') \
            .prefetch_related('tellers__user') \

    if request.user.is_staff:
        running_stories = all_running_stories
    else:
        running_stories = all_running_stories.filter(tellers__user=request.user)

    user_activity = User.objects \
            .filter(is_staff=False) \
            .annotate(parts_written=Count('teller__storyparts')) \
            .order_by('-parts_written', 'username')[:10]

    context = {
        'username': request.user.username,
        'specialpowers': request.user.is_staff,
        'running_stories': running_stories,
        'finished_stories': finished_stories,
        'user_activity': user_activity,
    }
    return render(request, 'umklapp/running.html', context)

def places(self, request, *args, **kwargs):
        snippet = self.get_object()
        lat = request.query_params['lat']
        lng = request.query_params['lng']

        userId = kwargs.get("id")
        listPlaces = []

        user = User.objects(id=userId).get()
        userTypes = user.types
        if not userTypes:
            userTypes = ['594c36a033cbb6eed6f364e6','594c36b233cbb6eed6f364f2','594c368633cbb6eed6f364da','594c4a504362de3d8fae464f','5956fee67d2bf938c292e0bc']

        places =  Place.objects(Q(loc__near=[float(lat), float(lng)]))

        for type in userTypes:
            typeName = Category.objects(id=type).get().name


            for place in places:
                if typeName in place.types:

                    # Rating place according to twitter
                    twitterRating = len(Tweet.objects(text__contains= place.name))
                    place.twitter_rating = twitterRating

                    #Rating place according to user
                    userRating = RatedPlace.objects(Q(user_id__exact=userId) & Q(place_id__exact=place.id) )
                    print userId
                    print place.id
                    print userRating

                    if userRating:

                        rating = userRating.order_by('-id').first().rating
                    else:
                        rating = ''
                    place.user_rating = rating

                    #Adding places to list of recommended places
                    listPlaces.append(place)

                    # Sort the list according to twitter rated
                    listPlaces.sort(key=lambda x: x.twitter_rating, reverse=True)

                    # Sort the list according to user rated
                    listPlaces.sort(key=lambda x: x.user_rating, reverse=True)

        serializer = PlaceSerializer(listPlaces[:20], many=True)

        return Response(serializer.data)

    ########User choosen categories

def semester_view(request, study_program, main_profile=None, semester_number=None, save_location=True):
    """
    Generates the link portal for a given semester in a given program code
    """
    try:
        semester = Semester.get(study_program, main_profile, semester_number)
    except Semester.DoesNotExist:
        if not main_profile and not semester_number:
            # This URL might refer to a userpage instead
            return studentpage(request, study_program)
        raise Http404(
            _('%s. semester ved hovedprofilen "%s" knyttet til studieprogrammet "%s" eksisterer ikke')
            % ((semester_number or 'Et'), main_profile or 'felles', study_program)
        )

    if save_location:
        # Save the deliberate change of location by user in the session, as the semester has been found successfully
        request.session['semester_pk'] = semester.pk
        request.session['study_program_slug'] = semester.study_program.slug

        # Delete studentpage slug in order to prevent redirection to it
        request.session['homepage'] = ''

        if semester.main_profile:
            request.session['main_profile_slug'] = semester.main_profile.slug
        else:
            request.session['main_profile_slug'] = ''
        request.session['semester_number_slug'] = semester.number
        request.session['homepage'] = ''  # Delete saved homepage

    # Where to send users if the semester has electives
    electives_url = ''
    if semester.has_electives:
        if request.user.is_authenticated:
            electives_url = request.user.options.get_admin_url()
        else:
            electives_url = '/accounts/dataporten/login'

    return render(
        request,
        'semesterpage/courses.html',
        {
            'semester': semester,
            'courses': semester.courses.all(),
            'study_programs': StudyProgram.objects.filter(published=True),
            'calendar_name': get_calendar_name(request),
            'electives_url': electives_url,
            'user': request.user,
            'student_page': False,
        },
    )