zulip/zerver/views/realm_domains.py

from django.core.exceptions import ValidationError
from django.http import HttpRequest, HttpResponse
from django.utils.translation import ugettext as _

from zerver.decorator import has_request_variables, require_realm_admin, REQ
from zerver.lib.actions import do_add_realm_domain, do_change_realm_domain, \
    do_remove_realm_domain
from zerver.lib.domains import validate_domain
from zerver.lib.response import json_error, json_success
from zerver.lib.validator import check_bool, check_string
from zerver.models import RealmDomain, UserProfile, get_realm_domains

from typing import Text

def list_realm_domains(request: HttpRequest, user_profile: UserProfile) -> HttpResponse:
    domains = get_realm_domains(user_profile.realm)
    return json_success({'domains': domains})

@require_realm_admin
@has_request_variables
def create_realm_domain(request: HttpRequest, user_profile: UserProfile,
                        domain: Text=REQ(validator=check_string),
                        allow_subdomains: bool=REQ(validator=check_bool)) -> HttpResponse:
    domain = domain.strip().lower()
    try:
        validate_domain(domain)
    except ValidationError as e:
        return json_error(_('Invalid domain: {}').format(e.messages[0]))
    if RealmDomain.objects.filter(realm=user_profile.realm, domain=domain).exists():
        return json_error(_("The domain %(domain)s is already a part of your organization.") % {'domain': domain})
    realm_domain = do_add_realm_domain(user_profile.realm, domain, allow_subdomains)
    return json_success({'new_domain': [realm_domain.id, realm_domain.domain]})

@require_realm_admin
@has_request_variables
def patch_realm_domain(request: HttpRequest, user_profile: UserProfile, domain: Text,
                       allow_subdomains: bool=REQ(validator=check_bool)) -> HttpResponse:
    try:
        realm_domain = RealmDomain.objects.get(realm=user_profile.realm, domain=domain)
        do_change_realm_domain(realm_domain, allow_subdomains)
    except RealmDomain.DoesNotExist:
        return json_error(_('No entry found for domain %(domain)s.' % {'domain': domain}))
    return json_success()

@require_realm_admin
@has_request_variables
def delete_realm_domain(request: HttpRequest, user_profile: UserProfile,
                        domain: Text) -> HttpResponse:
    try:
        realm_domain = RealmDomain.objects.get(realm=user_profile.realm, domain=domain)
        do_remove_realm_domain(realm_domain)
    except RealmDomain.DoesNotExist:
        return json_error(_('No entry found for domain %(domain)s.' % {'domain': domain}))
    return json_success()