Na}{

File Content: `subdomain.py`

from parallels.core.hosting_repository.subdomain import SubdomainModel, SubdomainEntity
from parallels.core.utils.migrator_utils import normalize_domain_name
from parallels.core.utils.common.string_utils import safe_idn_encode
from parallels.core.utils.mysql import escape_args_list
from parallels.plesk.hosting_repository.base import PleskBaseModel
from parallels.plesk.hosting_repository.utils.cli.domain import SubdomainCreateCli, DomainSetPhysicalHosting, \
    DomainSetForwarding


class PleskSubdomainModel(SubdomainModel, PleskBaseModel):
    def get_list(self, filter_subscription_name=None, filter_name=None):
        """Retrieve list of subdomains in target panel

        :type filter_subscription_name: str | unicode | None
        :type filter_name: list[str] | None
        :rtype: list[parallels.core.hosting_repository.subdomain.SubdomainEntity]
        """
        subdomains = []

        query = """
            SELECT
                domains.id AS subdomain_id,
                domains.name AS subdomain_name,
                parent_domain.name AS subscription_name
            FROM
                domains LEFT JOIN domains AS parent_domain ON parent_domain.id = domains.webspace_id
            WHERE
                domains.parentDomainId != 0
        """
        query_args = {}
        if filter_name is not None:
            if len(filter_name) == 0:
                return []

            filter_name_placeholders, filter_name_values = escape_args_list(filter_name, 'name')
            query += ' AND LOWER(domains.name) in ({filter_name_placeholders_str})'.format(
                filter_name_placeholders_str=', '.join(filter_name_placeholders)
            )
            # idn domains stored in database in idna encoding, so perform encoding of given filter
            query_args.update({key: normalize_domain_name(value) for key, value in filter_name_values.items()})

        for row in self.plesk_server.query_panel_db(query, query_args):
            subdomains.append(SubdomainEntity(
                subdomain_id=row['subdomain_id'],
                name=row['subdomain_name'],
                subscription_name=row['subscription_name']
            ))

        return subdomains

    def create_from_dump(self, subdomain_dump, domain_name, subscription_name):
        """Create subdomain in target Plesk

        :type subdomain_dump: parallels.core.dump.data_model.Subdomain
        :type domain_name: str
        :type subscription_name: str
        """
        command = SubdomainCreateCli(self.plesk_cli_runner, subdomain_dump, domain_name, subscription_name)
        command.run()

    def set_forwarding(self, subscription_name, subdomain_name, forwarding_url, is_frame_forwarding):
        """Enable forwarding on subdomain with given name in target Plesk

        :type subscription_name: str
        :type subdomain_name: str
        :type forwarding_url: str
        :type is_frame_forwarding: bool
        """
        command = DomainSetForwarding(
            self.plesk_cli_runner, subscription_name, subdomain_name, forwarding_url, is_frame_forwarding
        )
        command.run()

    def set_physical_hosting(self, subscription_name, subdomain_name, document_root, is_enable_ssl):
        """Enable physical hosting on subdomain with given name in target Plesk

        :type subscription_name: str
        :type subdomain_name: str
        :type document_root: str
        :type is_enable_ssl: bool
        """
        command = DomainSetPhysicalHosting(
            self.plesk_cli_runner, subscription_name, subdomain_name, document_root, is_enable_ssl
        )
        command.run()

    def is_exists_by_subscription(self, subscription_name, subdomain_name, guid=None):
        """Check if subdomain with given name or guid exists in target Plesk

        :type subscription_name: str
        :type subdomain_name: str
        :type guid: str | None
        :rtype: bool
        """
        if not guid:
            result = self.plesk_server.query_panel_db(
                'SELECT id FROM domains WHERE name = %(subdomain_name)s', dict(
                    subdomain_name=safe_idn_encode(subdomain_name)
                )
            )
        else:
            result = self.plesk_server.query_panel_db(
                'SELECT id FROM domains WHERE name = %(subdomain_name)s or guid = %(guid)s', dict(
                    subdomain_name=safe_idn_encode(subdomain_name),
                    guid=guid
                )
            )
        return len(result) > 0

Name	Size	Permission
__pycache__	---	0770
utils	---	0755
__init__.py	0 bytes	0644
base.py	772 bytes	0644
customer.py	7865 bytes	0644
database.py	4056 bytes	0644
database_user.py	3265 bytes	0644
dns_record.py	5991 bytes	0644
domain.py	4358 bytes	0644
domain_alias.py	2333 bytes	0644
extension.py	6333 bytes	0644
mail_account.py	8812 bytes	0644
model.py	5869 bytes	0644
panel.py	7158 bytes	0644
panel_user.py	2408 bytes	0644
panel_user_role.py	639 bytes	0644
reseller.py	9650 bytes	0644
service_plan.py	5139 bytes	0644
service_plan_addon.py	4325 bytes	0644
subdomain.py	4526 bytes	0644
subscription.py	20236 bytes	0644
tomcat.py	2162 bytes	0644
virtual_hosting_settings.py	5403 bytes	0644

File Content: subdomain.py

File Content: `subdomain.py`