Skip to content

downloader

xsdata.utils.downloader

Downloader

Remote recursive resource downloader.

Helper class to download a schema or a definitions with all their imports locally. The imports paths will be adjusted if necessary.

Parameters:

Name Type Description Default
output Path

The output path

 required

Attributes:

Name Type Description
base_path

The base path for the resources
downloaded dict

A cache of the downloaded resources
Source code in xsdata/utils/downloader.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
class Downloader:
    """Remote recursive resource downloader.

    Helper class to download a schema or a definitions with all their imports
    locally. The imports paths will be adjusted if necessary.

    Args:
        output: The output path

    Attributes:
        base_path: The base path for the resources
        downloaded: A cache of the downloaded resources
    """

    __slots__ = ("base_path", "downloaded", "output")

    def __init__(self, output: Path):
        """Initialize the downloader."""
        self.output = output
        self.downloaded: dict = {}

    def wget(self, uri: str, location: str | None = None) -> None:
        """Download handler for any uri input with circular protection."""
        if uri not in self.downloaded:
            self.downloaded[uri] = None
            logger.info("Fetching %s", uri)

            input_stream = opener.open(uri).read()  # nosec
            if uri.endswith("wsdl"):
                self.parse_definitions(uri, input_stream)
            else:
                self.parse_schema(uri, input_stream)

            self.write_file(uri, location, input_stream.decode())

    def parse_schema(self, uri: str, content: bytes) -> None:
        """Convert content to a schema instance and process all sub imports."""
        parser = SchemaParser(location=uri)
        schema = parser.from_bytes(content, Schema)
        self.wget_included(schema)

    def parse_definitions(self, uri: str, content: bytes) -> None:
        """Convert content to a definitions instance and process all sub imports."""
        parser = DefinitionsParser(location=uri)
        definitions = parser.from_bytes(content, Definitions)
        self.wget_included(definitions)

        for schema in definitions.schemas:
            self.wget_included(schema)

    def wget_included(self, definition: Schema | Definitions) -> None:
        """Download the definitions included resources."""
        for included in definition.included():
            if included.location:
                schema_location = getattr(included, "schema_location", None)
                self.wget(included.location, schema_location)

    def write_file(self, uri: str, location: str | None, content: str) -> None:
        """Write the downloaded uri to a local file.

        Keep track of all the written file paths, in case we have to
        modify the location attribute in an upcoming schema/definition
        import.

        Args:
            uri: The resource URI
            location: The import location of the resource
            content: The raw content string
        """
        if uri.startswith("file:"):
            # This happens for bundled schemas (xlink.xsd, xml.xsd)
            # which are intercepted by opener and returned as file://
            if not location or location.startswith("file:"):
                raise ValueError(
                    f"Cannot download local file without HTTP location: {uri}\n"
                )

            parsed = urlparse(location)
        else:
            parsed = urlparse(uri)

        rel_path = parsed.netloc + "/" + parsed.path.lstrip("/")
        file_path = self.output.joinpath(rel_path)

        content = self.adjust_imports(file_path.parent, content)
        file_path.parent.mkdir(parents=True, exist_ok=True)
        file_path.write_text(content, encoding="utf-8")

        logger.info("Writing %s", file_path)
        self.downloaded[uri] = file_path

    def adjust_imports(self, path: Path, content: str) -> str:
        """Update the location of the imports to point to the downloaded files."""
        matches = re.findall(r"ocation=\"(.*)\"", content)
        for match in matches:
            if isinstance(self.downloaded.get(match), Path):
                location = os.path.relpath(self.downloaded[match], path)
                replace = location.replace("\\", "/")
                content = content.replace(f'ocation="{match}"', f'ocation="{replace}"')

        return content

__init__(output)

Initialize the downloader.

Source code in xsdata/utils/downloader.py 
29
30
31
32
def __init__(self, output: Path):
    """Initialize the downloader."""
    self.output = output
    self.downloaded: dict = {}

wget(uri, location=None)

Download handler for any uri input with circular protection.

Source code in xsdata/utils/downloader.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
def wget(self, uri: str, location: str | None = None) -> None:
    """Download handler for any uri input with circular protection."""
    if uri not in self.downloaded:
        self.downloaded[uri] = None
        logger.info("Fetching %s", uri)

        input_stream = opener.open(uri).read()  # nosec
        if uri.endswith("wsdl"):
            self.parse_definitions(uri, input_stream)
        else:
            self.parse_schema(uri, input_stream)

        self.write_file(uri, location, input_stream.decode())

parse_schema(uri, content)

Convert content to a schema instance and process all sub imports.

Source code in xsdata/utils/downloader.py 
48
49
50
51
52
def parse_schema(self, uri: str, content: bytes) -> None:
    """Convert content to a schema instance and process all sub imports."""
    parser = SchemaParser(location=uri)
    schema = parser.from_bytes(content, Schema)
    self.wget_included(schema)

parse_definitions(uri, content)

Convert content to a definitions instance and process all sub imports.

Source code in xsdata/utils/downloader.py 
54
55
56
57
58
59
60
61
def parse_definitions(self, uri: str, content: bytes) -> None:
    """Convert content to a definitions instance and process all sub imports."""
    parser = DefinitionsParser(location=uri)
    definitions = parser.from_bytes(content, Definitions)
    self.wget_included(definitions)

    for schema in definitions.schemas:
        self.wget_included(schema)

wget_included(definition)

Download the definitions included resources.

Source code in xsdata/utils/downloader.py 
63
64
65
66
67
68
def wget_included(self, definition: Schema | Definitions) -> None:
    """Download the definitions included resources."""
    for included in definition.included():
        if included.location:
            schema_location = getattr(included, "schema_location", None)
            self.wget(included.location, schema_location)

write_file(uri, location, content)

Write the downloaded uri to a local file.

Keep track of all the written file paths, in case we have to modify the location attribute in an upcoming schema/definition import.

Parameters:

Name Type Description Default
uri str

The resource URI

 required
location str | None

The import location of the resource

 required
content str

The raw content string

 required
Source code in xsdata/utils/downloader.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def write_file(self, uri: str, location: str | None, content: str) -> None:
    """Write the downloaded uri to a local file.

    Keep track of all the written file paths, in case we have to
    modify the location attribute in an upcoming schema/definition
    import.

    Args:
        uri: The resource URI
        location: The import location of the resource
        content: The raw content string
    """
    if uri.startswith("file:"):
        # This happens for bundled schemas (xlink.xsd, xml.xsd)
        # which are intercepted by opener and returned as file://
        if not location or location.startswith("file:"):
            raise ValueError(
                f"Cannot download local file without HTTP location: {uri}\n"
            )

        parsed = urlparse(location)
    else:
        parsed = urlparse(uri)

    rel_path = parsed.netloc + "/" + parsed.path.lstrip("/")
    file_path = self.output.joinpath(rel_path)

    content = self.adjust_imports(file_path.parent, content)
    file_path.parent.mkdir(parents=True, exist_ok=True)
    file_path.write_text(content, encoding="utf-8")

    logger.info("Writing %s", file_path)
    self.downloaded[uri] = file_path

adjust_imports(path, content)

Update the location of the imports to point to the downloaded files.

Source code in xsdata/utils/downloader.py 
104
105
106
107
108
109
110
111
112
113
def adjust_imports(self, path: Path, content: str) -> str:
    """Update the location of the imports to point to the downloaded files."""
    matches = re.findall(r"ocation=\"(.*)\"", content)
    for match in matches:
        if isinstance(self.downloaded.get(match), Path):
            location = os.path.relpath(self.downloaded[match], path)
            replace = location.replace("\\", "/")
            content = content.replace(f'ocation="{match}"', f'ocation="{replace}"')

    return content