Quickstart

Five minutes to your first DataFrame.

Install PySUS:

pip install pysus

The recommended way to fetch data is through an origin namespace. Each origin exposes the same per-database fetchers, and the data source is explicit: :

import pysus

# DATASUS FTP (served from the S3 catalog mirror by default)
df = pysus.ftp.sinan(disease="deng", year=2024, as_dataframe=True)

# dados.gov.br (CKAN)
df = pysus.dadosgov.sinasc(state="SP", year=2024, as_dataframe=True)

# dadosabertos.saude.gov.br (theme datasets)
df = pysus.saude.arboviroses(year=2024, as_dataframe=True)

source="catalog" (default) serves the S3/Parquet mirror; pass source="origin" to query the origin server directly. The legacy flat functions (pysus.sinan, …) still work but emit a deprecation warning pointing to the namespaced form.

What you get back

Namespaced fetchers return either a FileBag or a DataFrame:

  • download=False → a remote FileBag listing the files that would be fetched, without downloading anything (as_dataframe is ignored here); call bag.download(), bag.download_one(i) or bag[i].download() to fetch.

  • download=True (default) + as_dataframe=False → a local FileBag of downloaded files.

  • as_dataframe=True → a single concatenated pandas.DataFrame.

A FileBag is a high-level, synchronous container. Its repr lists each file (remote files are marked (remote)); len/iter/[]/paths let you inspect it, and to_dataframe()/df concatenates local tabular files:

bag = pysus.saude.arboviroses(download=False)
# Files[fa_casoshumanos_1994-2026.csv (remote), fa_epizpnh_1994-2026.csv (remote)]
local = bag.download()          # -> FileBag of downloaded local files
df = local.to_dataframe()       # -> concatenated pandas.DataFrame

The legacy flat fetchers (pysus.sinan, …) keep their historic list[str] | pd.DataFrame return type.

S3 catalog (DuckLake) — the shared mirror

Every origin namespace reads from the DuckLake/S3 catalog by default. For lower-level control of that catalog, use the PySUS class:

import asyncio
from pysus.api.client import PySUS

async def main():
    async with PySUS() as pysus:
        files = await pysus.query(
            dataset="sinan",
            group="DENG",
            year=2024,
        )
        for f in files:
            local = await pysus.download(f)
            print(local.path)

asyncio.run(main())

FTP DATASUS

import asyncio
from pysus.api.ftp.client import FTP

async def main():
    ftp = FTP()
    await ftp.connect()
    try:
        for dataset in await ftp.datasets():
            if dataset.name != "SINAN":
                continue
            for item in await dataset.content:
                for file in await item.files:
                    if file.year == 2024 and "DENG" in file.name:
                        local = await file.download()
                        print(local.path)
    finally:
        await ftp.close()

asyncio.run(main())

DadosGov API

DadosGov requires an API token (see Environment Variables):

import asyncio
from pysus.api.dadosgov.client import DadosGov

async def main():
    client = DadosGov()
    await client.connect(token="<your-token>")
    try:
        for dataset in await client.datasets():
            for group in await dataset.content:
                for file in await group.files:
                    if file.year == 2024:
                        local = await file.download()
                        print(local.path)
    finally:
        await client.close()

 asyncio.run(main())

OpenDataSUS (Saude)

Public access — no token required. The simplest way to query the Ministry of Health’s open-data catalog:

import pysus

# Dengue notifications from OpenDataSUS
df = pysus.saude.arboviroses(disease="dengue", year=2024)

# Vaccination coverage
df = pysus.saude.vacinacao(state="SP", year=2024)

Or use the Saude client directly:

import asyncio
from pysus.api.saude import SaudeClient

async def main():
    async with SaudeClient() as client:
        page = await client.list_datasets(group="arboviroses")
        for entry in page:
            print(entry.name, entry.title)

asyncio.run(main())

Next steps