DLTINS Data Processing Pipeline

A modular and extensible Python pipeline for processing DLTINS data.

Features

• Parse XML metadata files to extract DLTINS download links
• Download DLTINS ZIP archives and extract XML content
• Convert DLTINS XML content into structured CSV format
• Enrich data by adding custom analysis columns
• Export data to local or remote storage systems using fsspec

Requirements

• Python 3.10+
• pandas
• requests
• lxml
• fsspec
• pandas-stubs

To support specific cloud systems, install additional fsspec plugins as needed:

• Amazon S3: pip install s3fs
• Google Cloud Storage: pip install gcsfs
• Azure Blob Storage: pip install adlfs

Development Dependencies:

• pytest
• pytest-cov
• pytest-mock
• ruff
• mypy
• pre-commit
• types-requests

Development

1. Install dependencies

poetry install

2. Run tests

poetry run pytest --cov=dltins_processor --cov-report=xml --cov-report=term

Usage

1. Download the XML file from a given URL

Downloads XML content from a specified URL.

from dltins_processor import XMLDownloader

xml_download = XMLDownloader('https://example.com')
xml_file = xml_download.download()

2. Parse the XML file and get the DLTINS Link

Parses an XML metadata file and extracts DLTINS download links.

from dltins_processor import DLTINSLinksParser

xml_parser = DLTINSLinksParser(xml_file)
link = xml_parser.get_specific_dltins_link(1)

3. Download and Extract DLTINS XML

Downloads and extracts a ZIP file containing the DLTINS XML from a given URL.

from dltins_processor import DLTINSDownloader

downloader = DLTINSDownloader(zip_link=link)
xml_content = downloader.extract_file()

4. Convert XML to CSV

Converts a DLTINS XML string into a CSV file and a DataFrame.

from dltins_processor import FileConverter

converter = FileConverter(xml_content, csv_dir="./output")
df = converter.convert_to_csv()

5. Add a_count Columns

Adds two columns to a DLTINS DataFrame:

• a_count: Number of lowercase 'a' characters in the FinInstrmGnlAttrbts.FullNm column.
• contains_a: 'YES' if a_count > 0, otherwise 'NO'.

from dltins_processor import AddACountsColumns

transformer = AddACountsColumns("./output", "./output")
transformer.add_a_counts_columns()

6. Upload to Cloud or Remote Filesystem

Saves a CSV file into a cloud or local file storage system using fsspec.

from dltins_processor import CloudStorage

storage = CloudStorage(
    csv_dir="./output",
    system="file",  # or "s3", "gcs", etc.
    storage_path="./backup/transformed_dltins_file.csv"
)
storage.run()

Author

Alexandre Oliveira – @ao60337