API Reference

The pysus.api package provides a layered architecture for discovering, downloading, and reading data from Brazilian public health databases (DATASUS). It supports four remote data sources: FTP DataSUS, dados.gov.br, OpenDataSUS (dadosabertos.saude.gov.br), and the DuckLake/S3 catalog mirror.

Architecture Overview

The package is organized into a hierarchy of abstract base classes and concrete implementations:

pysus/api/
├── __init__.py          # Package entry (re-exports PySUS)
├── client.py            # Main PySUS orchestrator
├── extensions.py        # File format handlers
├── models.py            # Abstract base classes
├── types.py             # Type aliases
├── _impl/
│   └── databases.py     # High-level convenience functions
├── ducklake/            # S3 DuckLake catalog client
├── ftp/                 # FTP client
├── dadosgov/            # dados.gov.br API client
├── saude/               # OpenDataSUS catalog client
├── quality/             # Data quality and profiling
├── transform/           # Data transformation
├── export/              # Export to CSV/Excel/GeoJSON/SQL
└── diff/                # DataFrame comparison

Quick Start

The simplest way to use PySUS is through an origin namespace, which makes the data source explicit:

import pysus

df = pysus.ftp.sinan(disease="dengue", year=2023)
df = pysus.saude.arboviroses(disease="dengue", year=2023)

Or with the async API:

from pysus.api.client import PySUS

async with PySUS() as pysus:
    files = await pysus.query(dataset="sinan", group="DENG", year=2023)
    for f in files:
        await pysus.download(f)

FileBag

High-level file-collection entity for the public API.

FileBag wraps either a list of BaseRemoteFile (nothing downloaded yet) or a list of BaseLocalFile (already local, e.g. Parquet, CSV, DBC) into a single, typed container.

The bag is a high-level surface: synchronous download() hides the async machinery of the underlying PySUS client. It is returned by the origin-namespaced fetchers (pysus.ftp.*, pysus.dadosgov.*, pysus.saude.*); the flat, deprecated fetchers keep their historic list[str] | pd.DataFrame return type.

class pysus.api.bag.FileBag(files: list[F] | tuple[F, ...])

Bases: Generic[F]

A high-level collection of one or more files.

F is either BaseRemoteFile (files are not downloaded yet) or BaseLocalFile (files are already on disk). All public methods are synchronous; any async download machinery is hidden inside the bag.

property df: pd.DataFrame

Alias for to_dataframe() (concatenated local frames).

download(indexes: list[int] | tuple[int, ...] | None = None) → FileBag[BaseLocalFile]

Download the (remote) files and return a local FileBag.

If the bag already holds local files this is a no-op and returns self. indexes optionally selects a subset of files (defaults to all). Runs synchronously; the async PySUS download machinery is started and awaited internally.

download_one(index: int = 0) → BaseLocalFile

Download a single file and return the local file entity.

property files: tuple[F, ...]

The underlying file entities, frozen as a tuple.

property first: F

The first file entity.

property kind: str

"remote" for remote files, "local" otherwise.

property paths: list[str]

Local/repo cache paths, or remote keys if still remote.

to_dataframe() → pd.DataFrame

Concatenate all local tabular files into one DataFrame.

Only meaningful when the bag holds local files; remote files must be downloaded first.

Main Client

Main orchestrator for the PySUS data pipeline.

Manages file downloads, local state tracking, catalog attachment, Parquet conversion, and query execution across multiple backends.

Supports both asynchronous (async with PySUS() as c:) and synchronous (with PySUS() as c:) usage. When entered synchronously, every async method automatically runs on a persistent event loop so callers never need async/await.

class pysus.api.client.Base(**kwargs: Any)

Bases: DeclarativeBase

Base declarative class for SQLAlchemy ORM models.

metadata: ClassVar[MetaData] = MetaData()

Refers to the _schema.MetaData collection that will be used for new _schema.Table objects.

See also

Accessing Table and Metadata

registry: ClassVar[_RegistryType] = <sqlalchemy.orm.decl_api.registry object>

Refers to the _orm.registry in use where new _orm.Mapper objects will be associated.

class pysus.api.client.DownloadStatus(value)

Bases: Enum

Download status values tracked for each local file.

COMPLETED = 'completed'

DOWNLOADING = 'downloading'

FAILED = 'failed'

MISSING = 'missing'

PENDING = 'pending'

class pysus.api.client.LocalFileState(**kwargs)

Bases: Base

ORM model tracking the state of a downloaded local file.

client_name: Mapped[str]

group: Mapped[str | None]

last_synced: Mapped[datetime]

month: Mapped[int | None]

path: Mapped[str]

remote_path: Mapped[str]

sha256: Mapped[str | None]

state: Mapped[str | None]

status: Mapped[DownloadStatus]

year: Mapped[int | None]

class pysus.api.client.PySUS(db_path: Path = PosixPath('/home/docs/pysus/config.db'))

Bases: object

Central orchestrator for downloading and querying PySUS datasets.

download(file: BaseRemoteFile, token: str | None = None, callback: Callable | None = None, timeout: float | None = None) → BaseLocalFile

Download a remote file and return a local file handle.

Skips re-download if a matching local copy already exists. When no callback is supplied and progress bars are enabled, a tqdm bar is shown automatically.

Parameters:

• file (BaseRemoteFile) – The remote file to download.

• token (str, optional) – Access token for authenticated clients (e.g. DadosGov).

• callback (Callable, optional) – Progress callback invoked during the download.

• timeout (float, optional) – Maximum seconds to wait for the download. None (default) means no timeout.

Returns:

The downloaded file wrapped in the appropriate handler.

Return type:

BaseLocalFile

Raises:

• ValueError – If the file’s client is not recognised.

• RuntimeError – If the download fails for any reason.

download_to_parquet(file: BaseRemoteFile, token: str | None = None, callback: Callable[[int, int], None] | None = None, timeout: float | None = None, add_dv: bool = True) → Parquet

Download a file and convert it to Parquet format.

Parameters:

• file (BaseRemoteFile) – The remote file to download and convert.

• token (str, optional) – Access token for authenticated clients.

• callback (Callable[[int, int], None], optional) – Progress callback.

• timeout (float, optional) – Maximum seconds to wait for the download.

• add_dv (bool, optional) – Whether to apply the IBGE verification digit on load (default True).

Returns:

The converted Parquet file handler.

Return type:

Parquet

Raises:

NotImplementedError – If the downloaded file type cannot be converted to Parquet.

get_completed_remote_paths() → set[str]

Return remote paths for all successfully downloaded files.

get_dadosgov(access_token: str | None) → DadosGov

Return the DadosGov client, connecting lazily if needed.

get_ducklake(callback: Callable[[int, int], None] | None = None) → DuckLake

Return the DuckLake client, initializing it lazily if needed.

get_ftp() → FTP

Return the FTP client, connecting lazily if needed.

get_local_file(file: BaseRemoteFile) → BaseLocalFile | None

Look up a previously downloaded file by its remote path.

get_local_hierarchy()

Build a nested dict of cached files grouped by client and dataset.

Returns:

Nested dict keyed by {client: {dataset: {group: [files]}}}.

Return type:

dict

get_saude() → SaudeClient

Return the Saude (dadosabertos.saude.gov.br) client.

No authentication is required; the client is lazily created.

query(client: Annotated[str, AfterValidator(func=_validate_origin)] | None = None, dataset: str | list[str] | None = None, group: str | list[str] | None = None, state: str | list[str] | None = None, year: int | list[int] | None = None, month: int | list[int] | None = None, as_dataframe: bool = False, columns: list[str] | None = None, dtypes: dict[str, str] | None = None) → list[BaseRemoteFile] | DataFrame

Query available datasets through the DuckLake catalog.

Parameters:

• client (Origin, optional) – Source client to filter by.

• dataset (str or list of str, optional) – Dataset name(s) to filter by.

• group (str or list of str, optional) – Group name pattern(s) to filter by (case-insensitive ILIKE).

• state (str or list of str, optional) – Two-letter state code(s) to filter by.

• year (int or list of int, optional) – Year(s) to filter by.

• month (int or list of int, optional) – Month(s) to filter by.

• as_dataframe (bool, optional) – When True, downloads all matching files, converts them to Parquet, and returns a single concatenated DataFrame. Default False.

• columns (list of str, optional) – Subset of column names to keep when as_dataframe=True.

• dtypes (dict, optional) – Override column types when as_dataframe=True. Keys are column names, values are dtype strings.

Returns:

List of matching File objects, or a DataFrame if as_dataframe=True.

Return type:

list or DataFrame

read_parquet(paths: list[Path], sql: str | None = None, mode: Literal['union', 'intersection', 'strict'] = 'union', add_dv: bool = True) → DuckDBPyConnection | pd.DataFrame

Read Parquet files with optional schema handling and SQL filter.

Parameters:

• paths (list of Path) – One or more Parquet file paths to read.

• sql (str, optional) – Optional SQL filter expression applied to the result.

• mode ({"union", "intersection", "strict"}, optional) – Schema resolution mode (default "union").

• add_dv (bool, optional) – When True, automatically applies the IBGE verification digit to municipality code columns. If matching columns are found, a DataFrame is returned instead of a DuckDBPyConnection.

Returns:

The query result.

Return type:

DuckDBPyConnection or pd.DataFrame

Raises:

ValueError – If no paths are provided, or if the schema mode is "strict" and the files have differing schemas.

Types

Utilities

pysus.api.utils.add_dv(geocode: str) → str

Add the IBGE verification digit to a municipality code.

Parameters:

geocode (str) – The municipality code (6 or 7 digits).

Returns:

The code with the verification digit appended, or the original string if it cannot be processed.

Return type:

str

pysus.api.utils.is_geocode_column(name: str) → bool

Check if a column name corresponds to an IBGE municipality code.

File Format Handlers

Map file extensions and MIME types to their handler classes.

class pysus.api.extensions.CSV(*, path: Path, type: Annotated[str, AfterValidator(func=_validate_file_type)] = 'CSV')

Bases: BaseTabularFile

Represents a CSV file with automatic encoding and separator detection.

property columns: list[Column]

Return the column metadata from the CSV header row.

async load() → DataFrame

Read the entire CSV into a DataFrame.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property rows: int

Return the number of data rows in the file.

async stream(chunk_size: int = 10000) → AsyncGenerator[DataFrame, None]

Yield the CSV in chunks of the given number of rows.

type: Annotated[str, AfterValidator(func=_validate_file_type)]

class pysus.api.extensions.DBC(*, path: Path, type: Annotated[str, AfterValidator(func=_validate_file_type)] = 'DBC')

Bases: BaseTabularFile

Represents a compressed DBC file, convertible to DBF then Parquet.

property columns: list[Column]

Not supported for DBC files. Convert to Parquet first.

async load() → DataFrame

Convert to Parquet and load the result as a DataFrame.

property rows: int

Not supported for DBC files. Convert to Parquet first.

async stream(chunk_size: int = 10000) → AsyncGenerator[DataFrame, None]

Convert to Parquet and stream its chunks.

async to_parquet(output_path: str | Path | None = None, chunk_size: int = 30000, callback: Callable[[int, int], None] | None = None) → Parquet

Convert the file to Parquet format.

Parameters:

• output_path (str or Path, optional) – Destination path for the Parquet file. Defaults to the source path with a .parquet extension.

• chunk_size (int, optional) – Number of rows per streaming chunk (default 10 000).

• callback (Callable[[int, int], None], optional) – Function called after each chunk with (current_rows, total_rows).

Returns:

The resulting Parquet wrapper object.

Return type:

Parquet

type: Annotated[str, AfterValidator(func=_validate_file_type)]

class pysus.api.extensions.DBF(*, path: Path, type: Annotated[str, AfterValidator(func=_validate_file_type)] = 'DBF')

Bases: BaseTabularFile

Represents a dBASE (DBF) file.

property columns: list[Column]

Return the column metadata from the DBF file.

decode_column(value)

Decode a raw DBF value, handling byte strings and null bytes.

Parameters:

value (bytes or str or Any) – The value to decode.

Returns:

The decoded and stripped string, or the original value if it is neither bytes nor str.

Return type:

str or Any

async load(fast: bool = True) → DataFrame

Read the entire DBF file into a DataFrame.

Parameters:

fast (bool) – If True use the byte-level reader (default), falling back to dbfread on failure. If False use dbfread.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property rows: int

Return the number of records in the DBF file.

async stream(chunk_size: int = 30000, fast: bool = True) → AsyncGenerator[DataFrame, None]

Yield the DBF records in chunks of the given size.

Parameters:

• chunk_size (int) – Number of rows per chunk.

• fast (bool) – If True use the byte-level reader (default), falling back to dbfread on failure. If False use dbfread.

async to_parquet(output_path: str | Path | None = None, chunk_size: int = 30000, callback: Callable[[int, int], None] | None = None, fast: bool = True) → Parquet

Convert the DBF file to Parquet format.

Parameters:

• output_path (str or Path, optional)

• chunk_size (int) – Rows per chunk when building Parquet.

• callback (callable, optional)

• fast (bool) – If True use the byte-level reader (default). If False use dbfread.

type: Annotated[str, AfterValidator(func=_validate_file_type)]

class pysus.api.extensions.Directory(*, path: Path, type: Annotated[str, AfterValidator(func=_validate_file_type)] = 'DIR')

Bases: BaseLocalFile

Represents a directory on the local filesystem.

async load() → list[BaseLocalFile]

Load all entries inside the directory as file objects.

async stream(chunksize: int = 10000) → AsyncGenerator[BaseLocalFile, None]

Yield each entry inside the directory as a file object.

type: Annotated[str, AfterValidator(func=_validate_file_type)]

class pysus.api.extensions.ExtensionFactory

Bases: object

Factory that maps file content and extensions to handler classes.

async classmethod get_file_class(path: Path) → type[BaseLocalFile]

Return the handler class for a given path.

First attempts content-based identification; falls back to extension matching.

Parameters:

path (Path) – The file path to classify.

Returns:

The handler class for the file type.

Return type:

type[BaseLocalFile]

async classmethod instantiate(path: str | Path) → BaseLocalFile

Create and return the appropriate file handler for a path.

Determines whether the path is a directory or a file, resolves the handler class, and instantiates it.

Parameters:

path (str or Path) – The filesystem path to wrap in a handler.

Returns:

The instantiated file handler.

Return type:

BaseLocalFile

class pysus.api.extensions.File(*, path: Path, type: Annotated[str, AfterValidator(func=_validate_file_type)] = 'FILE')

Bases: BaseLocalFile

Represents a generic local file with no special handling.

async load() → bytes

Read the entire file contents into memory as bytes.

async stream(chunk_size: int = 1048576) → AsyncGenerator[bytes, None]

Yield the file contents in chunks of the given size.

type: Annotated[str, AfterValidator(func=_validate_file_type)]

class pysus.api.extensions.GZip(*, path: Path, type: Annotated[str, AfterValidator(func=_validate_file_type)] = 'GZIP')

Bases: BaseCompressedFile

Represents a GZip-compressed file.

async extract(target_dir: Path = PosixPath('/home/docs/pysus')) → list[BaseLocalFile]

Decompress the file to a target directory and return it as a file object.

async list_members() → list[str]

Return a list containing the single decompressed file name.

async load() → bytes

Decompress and read the entire file contents into memory.

async open_member(member_name: str) → bytes

Read and return the decompressed file contents.

type: Annotated[str, AfterValidator(func=_validate_file_type)]

class pysus.api.extensions.JSON(*, path: Path, type: Annotated[str, AfterValidator(func=_validate_file_type)] = 'JSON')

Bases: BaseTabularFile

Represents a JSON file with tabular data.

property columns: list[Column]

Return the column metadata from the JSON file.

async load() → DataFrame

Read the entire JSON file into a DataFrame.

property rows: int

Return the number of rows in the JSON file.

async stream(chunk_size: int = 10000) → AsyncGenerator[DataFrame, None]

Yield the entire JSON file as a single DataFrame.

type: Annotated[str, AfterValidator(func=_validate_file_type)]

class pysus.api.extensions.JSONL(*, path: Path, type: Annotated[str, AfterValidator(func=_validate_file_type)] = 'JSONL')

Bases: BaseTabularFile

Represents a JSON Lines file — one JSON object per line.

Used by the Saude client to persist paginated DEMAS REST rows.

property columns: list[Column]

Return the column metadata from the first record.

async load() → DataFrame

Read the entire JSONL file into a DataFrame.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property rows: int

Return the number of JSON records in the file.

async stream(chunk_size: int = 10000) → AsyncGenerator[DataFrame, None]

Yield the JSONL records in batches of the given size.

type: Annotated[str, AfterValidator(func=_validate_file_type)]

class pysus.api.extensions.PDF(*, path: Path, type: Annotated[str, AfterValidator(func=_validate_file_type)] = 'PDF')

Bases: BaseLocalFile

Represents a PDF file.

async load() → bytes

Read the entire PDF file contents into memory as bytes.

async stream(chunk_size: int | None = None) → AsyncGenerator[bytes, None]

Yield the PDF file contents in chunks of the given size.

type: Annotated[str, AfterValidator(func=_validate_file_type)]

class pysus.api.extensions.Parquet(*, path: Path, type: Annotated[str, AfterValidator(func=_validate_file_type)] = 'PARQUET', add_dv: bool = True)

Bases: BaseTabularFile

Represents a Parquet file with optional date and integer type parsing.

add_dv: bool

property columns: list[Column]

Return the column metadata from the Parquet schema.

async load(parse: bool = True) → DataFrame

Read the entire Parquet file into a DataFrame.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

static parse_dftypes(df: DataFrame) → DataFrame

Convert known date and integer columns to their proper types.

property rows: int

Return the number of rows from the Parquet metadata.

property schema: Schema

Return the Parquet schema as a PyArrow Schema object.

async stream(chunk_size: int = 10000, parse: bool = False) → AsyncGenerator[DataFrame, None]

Yield the Parquet file in batches of the given size.

type: Annotated[str, AfterValidator(func=_validate_file_type)]

class pysus.api.extensions.Tar(*, path: Path, type: Annotated[str, AfterValidator(func=_validate_file_type)] = 'TAR')

Bases: BaseCompressedFile

Represents a Tar archive file.

async extract(target_dir: Path = PosixPath('/home/docs/pysus')) → list[BaseLocalFile]

Extract members to a target directory and return as file objects.

async list_members() → list[str]

Return the list of member names inside the archive.

async load() → TarFile

Open and return the tar archive.

async open_member(member_name: str) → bytes

Read and return the contents of a named archive member.

type: Annotated[str, AfterValidator(func=_validate_file_type)]

class pysus.api.extensions.Zip(*, path: Path, type: Annotated[str, AfterValidator(func=_validate_file_type)] = 'ZIP')

Bases: BaseCompressedFile

Represents a ZIP archive file.

async extract(target_dir: Path = PosixPath('/home/docs/pysus')) → list[BaseLocalFile]

Extract members to a target directory and return as file objects.

async list_members() → list[str]

Return the list of member names inside the archive.

async load() → ZipFile

Open and return the ZIP archive.

async open_member(member_name: str) → bytes

Read and return the contents of a named archive member.

async to_parquet(output_path: str | Path | None = None, chunk_size: int = 30000, callback: Callable[[int, int], None] | None = None) → Parquet

Extract the archive and convert the first tabular file to Parquet.

type: Annotated[str, AfterValidator(func=_validate_file_type)]

Abstract Base Models

Abstract model hierarchy for PySUS data access.

Provides abstract base classes for local and remote file handling, organized in a layered hierarchy: BaseFile -> BaseLocalFile -> BaseTabularFile / BaseCompressedFile for local files, and BaseFile -> BaseRemoteFile for remote files, alongside BaseRemoteObject -> BaseRemoteGroup / BaseRemoteDataset / BaseRemoteClient for remote data catalogs.

class pysus.api.models.BaseCompressedFile(*, path: Path, type: str | Annotated[str, AfterValidator(func=_validate_file_type)])

Bases: BaseLocalFile, ABC

Abstract base for a compressed archive file (e.g. .zip, .gz).

Subclasses must implement list_members, open_member, and extract.

abstractmethod async extract(target_dir: Path = PosixPath('/home/docs/pysus')) → list[BaseLocalFile]

Extract all members into target_dir and return the file objects.

abstractmethod async list_members() → list[str]

Return the list of member names inside the archive.

abstractmethod async open_member(member_name: str) → Any

Open and return a single archive member by name.

async stream(chunk_size: int | None = None) → AsyncGenerator[Any, None]

Yield each archive member as it is opened.

class pysus.api.models.BaseFile(*, path: Path, type: str | Annotated[str, AfterValidator(func=_validate_file_type)])

Bases: BaseModel, ABC

Abstract base for a single file, local or remote.

Subclasses must implement name, extension, size, and modify.

property basename: str

Return the file name from the path.

abstract property extension: str

Return the file extension string.

abstract property modify: datetime

Return the last modification timestamp.

abstract property name: str

Return the display name of the file.

path: Path

abstract property size: int

Return the file size in bytes.

type: str | FileType

class pysus.api.models.BaseLocalFile(*, path: Path, type: str | Annotated[str, AfterValidator(func=_validate_file_type)])

Bases: BaseFile, ABC

Abstract base for a file stored on the local filesystem.

Subclasses must implement load and stream.

property extension: str

Return the file extension from the local path.

async get_hash(algorithm: str = 'sha256', chunk_size: int = 1048576) → str

Compute the file’s hash digest.

Parameters:

• algorithm (str, optional) – The hash algorithm name (default "sha256").

• chunk_size (int, optional) – Read chunk size in bytes (default 1 MiB).

Returns:

The hex digest string.

Return type:

str

abstractmethod async load() → Any

Load the entire file content into memory and return it.

property modify: datetime

Return the last modification timestamp from the local filesystem.

property name: str

Return the file name from the path.

path: Path

property size: int

Return the file size in bytes from the local filesystem.

abstractmethod stream(chunk_size: int = 10000) → AsyncGenerator[Any, None]

Yield chunks of the file content as an async generator.

class pysus.api.models.BaseRemoteClient

Bases: BaseRemoteObject, ABC

Abstract base for a remote API client (e.g. FTP, HTTP).

Subclasses must implement connect, close, login, datasets, and _download_file.

abstractmethod async close() → None

Close the connection to the remote server.

abstractmethod async connect() → None

Establish a connection to the remote server.

abstractmethod async datasets(**kwargs) → list

Return a list of available datasets matching kwargs.

abstractmethod async download(file: BaseRemoteFile, output: Path, callback: Callable[[int, int], None] | None = None) → Path

Download a single file to output and return the local path.

abstractmethod async login(**kwargs) → None

Authenticate with the remote server using kwargs credentials.

class pysus.api.models.BaseRemoteDataset

Bases: BaseRemoteObject, SearchableMixin, ABC

Abstract base for a dataset containing groups and/or files.

Subclasses must implement _fetch_content.

client: BaseRemoteClient

property content: Sequence[BaseRemoteGroup | BaseRemoteFile]

Return the dataset content, fetching on first access.

group_definitions: dict[str, str]

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

async search(**kwargs) → list[BaseRemoteFile]

Recursively search groups and files by attribute kwargs.

Return matching file objects.

class pysus.api.models.BaseRemoteFile(*, path: Path, type: str | Annotated[str, AfterValidator(func=_validate_file_type)])

Bases: BaseFile, SearchableMixin, MetadataMixin, ABC

Abstract base for a file stored on a remote server.

Subclasses must implement _download. dataset and group link back to the containing objects.

property client: BaseRemoteClient

Return the remote client associated with this file.

dataset: BaseRemoteDataset

async download(output: str | Path | None = None, callback: Callable[[int, int], None] | None = None) → BaseLocalFile

Download the remote file to a local cache or output path.

Return the instantiated local file wrapper.

group: BaseRemoteGroup | None

property month: int | None

Return the month associated with the file, or None.

property name: str

Return the basename as the display name.

property state: Annotated[str, AfterValidator(func=_validate_state)] | None

Return the state associated with the file, or None.

property year: int | None

Return the year associated with the file, or None.

class pysus.api.models.BaseRemoteGroup

Bases: BaseRemoteObject, SearchableMixin, ABC

Abstract base for a named group of remote files within a dataset.

Subclasses must implement _fetch_files.

dataset: BaseRemoteDataset

property files: list[BaseRemoteFile]

Return all files in this group, fetching them on first access.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property parent: BaseRemoteDataset

Return the parent dataset.

async search(**kwargs) → list[BaseRemoteFile]

Filter files in this group by attribute kwargs.

Return matching file objects.

class pysus.api.models.BaseRemoteObject

Bases: MetadataMixin, BaseModel, ABC

Abstract base for a named remote entity with a description.

Subclasses must implement name, long_name, and description.

abstract property description: str

Return a textual description of the entity.

abstract property long_name: str

Return the long / human-readable name.

abstract property name: str

Return the short name of the remote entity.

class pysus.api.models.BaseTabularFile(*, path: Path, type: str | Annotated[str, AfterValidator(func=_validate_file_type)])

Bases: BaseLocalFile, ABC

Abstract base for a local tabular file (e.g. CSV, Parquet).

Subclasses must implement columns, rows, load, and stream.

abstract property columns: list[Column]

Return the list of column metadata.

abstractmethod async load() → DataFrame

Load the entire file into a pandas DataFrame.

property metadata: MetadataBag

Return structure/access metadata computed from the file.

Columns come from the concrete columns property; row count and size from the local filesystem. No network is involved.

abstract property rows: int

Return the number of data rows.

abstractmethod stream(chunk_size: int = 10000) → AsyncGenerator[DataFrame, None]

Yield pandas DataFrames in chunks as an async generator.

async to_parquet(output_path: str | Path | None = None, chunk_size: int = 10000, callback: Callable[[int, int], None] | None = None) → Parquet

Convert the file to Parquet format.

Parameters:

• output_path (str or Path, optional) – Destination path for the Parquet file. Defaults to the source path with a .parquet extension.

• chunk_size (int, optional) – Number of rows per streaming chunk (default 10 000).

• callback (Callable[[int, int], None], optional) – Function called after each chunk with (current_rows, total_rows).

Returns:

The resulting Parquet wrapper object.

Return type:

Parquet

class pysus.api.models.MetadataMixin

Bases: object

Mixin providing the .metadata / .ametadata() contract.

Shared by BaseRemoteFile and BaseRemoteObject so every remote entity (client, dataset, group, file) surfaces the same merged MetadataBag.

Concrete subclasses declare their extractors via extractor_types; the mixin only knows that some extractors exist and merges their output with the default cross-origin precedence.

async ametadata() → MetadataBag

Async variant of metadata() (may hit the network).

extractor_types: ClassVar[list[type[MetadataExtractor]]] = []

Extractor classes the concrete subclass wants applied. Each client module (ftp/dadosgov/ducklake/saude) overrides this.

property metadata: MetadataBag

Return the merged metadata bag for this entity.

Runs every registered extractor synchronously, merges the bags with the default cross-origin precedence and caches the result. Extractor failures are ignored (best-effort).

Returns:

The merged metadata bag.

Return type:

MetadataBag

class pysus.api.models.SearchableMixin

Bases: object

Mixin providing attribute-based filtering for remote objects.

High-Level Data Functions

High-level convenience functions for fetching Brazilian health data.

Each function wraps an asynchronous query/download pipeline and returns a pandas DataFrame. The available datasets cover disease notification (SINAN), vital statistics (SINASC, SIM), hospital admissions (SIH), ambulatory care (SIA), immunisation (PNI), census data (IBGE), health facilities (CNES), hospitalisation records (CIHA), COVID-19, and all Saude open-data themes.

pysus.api._impl.databases.arboviroses(**kwargs) → list[str] | DataFrame

Fetch arboviroses (dengue, chikungunya, zika, yellow fever) from Saude.

Parameters:

**kwargs – Forwarded to _fetch_data().

Examples

>>> pysus.arboviroses(as_dataframe=True)

pysus.api._impl.databases.assistencia_saude(**kwargs) → list[str] | DataFrame

Fetch hospital and health facility data from Saude.

Examples

>>> pysus.assistencia_saude(as_dataframe=True)

pysus.api._impl.databases.atencao_primaria(**kwargs) → list[str] | DataFrame

Fetch primary care data (Previne Brasil, SISAB) from Saude.

Examples

>>> pysus.atencao_primaria(as_dataframe=True)

pysus.api._impl.databases.bnafar(**kwargs) → list[str] | DataFrame

Fetch pharmaceutical assistance (Hórus stock) from Saude.

Examples

>>> pysus.bnafar(as_dataframe=True)

pysus.api._impl.databases.ciencia_tecnologia(**kwargs) → list[str] | DataFrame

Fetch science & technology data (Conitec, RIPSA) from Saude.

Examples

>>> pysus.ciencia_tecnologia(as_dataframe=True)

pysus.api._impl.databases.ciha(state: Annotated[str, AfterValidator(func=_validate_state)], year: int | list[int], month: int | list[int], group: str | None = 'CIHA', **kwargs) → list[str] | DataFrame

Fetch CIHA hospitalisation records for state, year, month.

CIHA provides hospitalisation records.

Parameters:

• state (str) – Two-letter state abbreviation (e.g. "RJ").

• year (int | list[int]) – Year or list of years to fetch.

• month (int | list[int]) – Month or list of months to fetch.

• group (str, optional) – Grouping code. Default is "CIHA".

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.ciha("RJ", 2020, 1, as_dataframe=True)

pysus.api._impl.databases.cnes(state: Annotated[str, AfterValidator(func=_validate_state)], year: int | list[int], month: int | list[int], group: str | None = None, **kwargs) → list[str] | DataFrame

Fetch CNES health facilities for a state, year, month.

CNES is the Brazilian registry of health-care facilities.

Parameters:

• state (str) – Two-letter state abbreviation (e.g. "RJ").

• year (int | list[int]) – Year or list of years to fetch.

• month (int | list[int]) – Month or list of months to fetch.

• group (str, optional) – Additional grouping code.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.cnes("RJ", 2020, 1, as_dataframe=True)

pysus.api._impl.databases.covid19(**kwargs) → list[str] | DataFrame

Fetch COVID-19 confirmed cases from dados.gov.br.

Parameters:

**kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.covid19(as_dataframe=True)

pysus.api._impl.databases.diagnosticos_tratamentos(**kwargs) → list[str] | DataFrame

Fetch diagnostics & treatment protocols (PCDT) from Saude.

Examples

>>> pysus.diagnosticos_tratamentos(as_dataframe=True)

pysus.api._impl.databases.economia_saude(**kwargs) → list[str] | DataFrame

Fetch health economics data (BPS, ApuraSUS, SIOPS) from Saude.

Examples

>>> pysus.economia_saude(as_dataframe=True)

pysus.api._impl.databases.educacao_saude(**kwargs) → list[str] | DataFrame

Fetch health education data (PVC) from Saude.

Examples

>>> pysus.educacao_saude(as_dataframe=True)

pysus.api._impl.databases.ibge(year: int | list[int], group: str | None = None, **kwargs) → list[str] | DataFrame

Fetch IBGE census data for given year(s).

IBGE provides census and demographic data.

Parameters:

• year (int | list[int]) – Year or list of years to fetch.

• group (str, optional) – Additional grouping code.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.ibge(2020, as_dataframe=True)

pysus.api._impl.databases.list_files(dataset: Annotated[str, AfterValidator(func=_validate_dataset_name)], client: Annotated[str, AfterValidator(func=_validate_origin)] | None = None, group: str | None = None, state: str | None = None, year: int | list[int] | None = None, month: int | list[int] | None = None, **kwargs) → DataFrame

List catalog files filtered by client, group, state, year, and month.

Queries the PySUS API metadata and returns a DataFrame with file data without downloading the actual files.

Parameters:

• dataset (str) – Dataset name (e.g. "SINAN").

• client (str, optional) – Data source client to query.

• group (str, optional) – Group or disease code to filter by.

• state (str, optional) – Two-letter state abbreviation.

• year (int | list[int], optional) – Year or list of years to filter by.

• month (int | list[int], optional) – Month or list of months to filter by.

• **kwargs – Forwarded to PySUS.query().

Returns:

Columns: name, path, dataset, group, year, month, state, modify.

Return type:

pd.DataFrame

Examples

>>> pysus.list_files("SINAN", year=2020, state="RJ")

pysus.api._impl.databases.macro_saude(**kwargs) → list[str] | DataFrame

Fetch macro-region and health region data from Saude.

Examples

>>> pysus.macro_saude(as_dataframe=True)

pysus.api._impl.databases.outros_temas(**kwargs) → list[str] | DataFrame

Fetch miscellaneous CED coordination data from Saude.

Examples

>>> pysus.outros_temas(as_dataframe=True)

pysus.api._impl.databases.ouvidoria(**kwargs) → list[str] | DataFrame

Fetch SUS ombudsman complaints from Saude.

Examples

>>> pysus.ouvidoria(as_dataframe=True)

pysus.api._impl.databases.pda(**kwargs) → list[str] | DataFrame

Fetch digital health and open data plan data from Saude.

Examples

>>> pysus.pda(as_dataframe=True)

pysus.api._impl.databases.pni(state: Annotated[str, AfterValidator(func=_validate_state)], year: int | list[int], group: str | None = None, **kwargs) → list[str] | DataFrame

Fetch PNI immunisation records for a given state, year(s).

PNI is the Brazilian national immunisation programme.

Parameters:

• state (str) – Two-letter state abbreviation (e.g. "RJ").

• year (int | list[int]) – Year or list of years to fetch.

• group (str, optional) – Additional grouping code.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.pni("RJ", 2020, as_dataframe=True)

pysus.api._impl.databases.prevencao_promocao(**kwargs) → list[str] | DataFrame

Fetch prevention & promotion (EPI distribution) from Saude.

Examples

>>> pysus.prevencao_promocao(as_dataframe=True)

pysus.api._impl.databases.saude_indigena(**kwargs) → list[str] | DataFrame

Fetch indigenous health data (Siasi/SasiSUS/Sesai) from Saude.

Examples

>>> pysus.saude_indigena(as_dataframe=True)

pysus.api._impl.databases.sia(state: Annotated[str, AfterValidator(func=_validate_state)], year: int | list[int], month: int | list[int], group: str | None = None, **kwargs) → list[str] | DataFrame

Fetch SIA ambulatory care for a state, year, month.

SIA is the Brazilian ambulatory care information system.

Parameters:

• state (str) – Two-letter state abbreviation (e.g. "RJ").

• year (int | list[int]) – Year or list of years to fetch.

• month (int | list[int]) – Month or list of months to fetch.

• group (str, optional) – Additional grouping code.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.sia("RJ", 2020, 1, as_dataframe=True)

pysus.api._impl.databases.sih(state: Annotated[str, AfterValidator(func=_validate_state)], year: int | list[int], month: int | list[int], group: str | None = None, **kwargs) → list[str] | DataFrame

Fetch SIH hospital admissions for a state, year, month.

SIH is the Brazilian hospital admission information system.

Parameters:

• state (str) – Two-letter state abbreviation (e.g. "RJ").

• year (int | list[int]) – Year or list of years to fetch.

• month (int | list[int]) – Month or list of months to fetch.

• group (str, optional) – Additional grouping code.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.sih("RJ", 2020, 1, as_dataframe=True)

pysus.api._impl.databases.sim(state: Annotated[str, AfterValidator(func=_validate_state)], year: int | list[int], group: str | None = None, **kwargs) → list[str] | DataFrame

Fetch SIM mortality records for a given state, year(s).

SIM is the Brazilian mortality information system.

Parameters:

• state (str) – Two-letter state abbreviation (e.g. "RJ").

• year (int | list[int]) – Year or list of years to fetch.

• group (str, optional) – Additional grouping code.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.sim("RJ", 2020, as_dataframe=True)

pysus.api._impl.databases.sinan(disease: Annotated[str, AfterValidator(func=_validate_dataset_name)], year: int | list[int], **kwargs) → list[str] | DataFrame

Fetch SINAN records for a given disease and year(s).

SINAN is the Brazilian notifiable-disease information system.

Parameters:

• disease (str) – Disease code (e.g. "DENG" for dengue, "ZIKA").

• year (int | list[int]) – Year or list of years to fetch.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.sinan("DENG", 2020, as_dataframe=True)

pysus.api._impl.databases.sinasc(state: Annotated[str, AfterValidator(func=_validate_state)], year: int | list[int], group: str | None = None, **kwargs) → list[str] | DataFrame

Fetch SINASC birth certificates for a given state, year(s).

SINASC is the Brazilian live birth information system.

Parameters:

• state (str) – Two-letter state abbreviation (e.g. "RJ").

• year (int | list[int]) – Year or list of years to fetch.

• group (str, optional) – Additional grouping code.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.sinasc("RJ", 2020, as_dataframe=True)

pysus.api._impl.databases.sisagua(**kwargs) → list[str] | DataFrame

Fetch water quality surveillance data from Saude.

Examples

>>> pysus.sisagua(as_dataframe=True)

pysus.api._impl.databases.sisvan(**kwargs) → list[str] | DataFrame

Fetch food & nutrition surveillance data from Saude.

Examples

>>> pysus.sisvan(as_dataframe=True)

pysus.api._impl.databases.vacinacao(**kwargs) → list[str] | DataFrame

Fetch vaccination data (PNI doses, ESAVI) from Saude.

Examples

>>> pysus.vacinacao(as_dataframe=True)

pysus.api._impl.databases.vigilancia_meio_ambiente(**kwargs) → list[str] | DataFrame

Fetch environmental surveillance (SRAG, SIM, mpox) from Saude.

Examples

>>> pysus.vigilancia_meio_ambiente(as_dataframe=True)

Column Metadata

Searchable column metadata across SUS health datasets.

Provides a unified interface to look up column names, types, bilingual descriptions, value categories and field characteristics across all available schema sources (saude endpoints and SINAN disease forms).

Examples

>>> from pysus.api.columns import search_columns, ColumnInfo
>>> results = search_columns("dengue", "notification")
>>> for col in results:
...     print(col.name, col.description)

class pysus.api.columns.ColumnInfo(name: str, description: str = '', description_en: str = '', dtype: str = '', dataset: str = '', endpoint: str = '', categories: str = '', characteristics: str = '', required: bool = False, format: str = '')

Metadata for a single column in a SUS dataset.

name

Column name as it appears in the data file (lowercase).

Type:

str

description

Human-readable description (Portuguese).

Type:

str

description_en

English description, or empty string if unavailable.

Type:

str

dtype

Data type string (e.g. "integer", "string", "VARCHAR(1)").

Type:

str

dataset

Parent dataset name (e.g. "arboviroses", "sinan").

Type:

str

endpoint

Sub-grouping within the dataset (e.g. "dengue", or the SINAN disease form such as "peste").

Type:

str

categories

Value codes of the field (e.g. "1-Sim 2-Não 9-Ignorado"), or empty when unavailable.

Type:

str

characteristics

Field rules and notes from the official dictionary (obligatory, dependencies, validation), or empty when unavailable.

Type:

str

required

Whether the field is mandatory on the notification form.

Type:

bool

format

Expected format hint (e.g. "YYYYMMDD"), if any.

Type:

str

categories: str

characteristics: str

dataset: str

description: str

description_en: str

dtype: str

endpoint: str

format: str

name: str

required: bool

pysus.api.columns.search_columns(dataset: str | None = None, query: str | None = None, *, endpoint: str | None = None) → list[ColumnInfo]

Search column metadata across all available schema sources.

Parameters:

• dataset (str, optional) – Restrict to a specific dataset name (e.g. "arboviroses" or "sinan"). When None, searches all datasets.

• query (str, optional) – Case-insensitive substring match against column name and descriptions (both PT and EN). When None, returns all columns for the given dataset.

• endpoint (str, optional) – Restrict to a specific endpoint within the dataset (e.g. "dengue", or a SINAN disease form such as "peste").

Returns:

Matching column metadata, sorted by name.

Return type:

list[ColumnInfo]

Examples

>>> search_columns("arboviroses", "date")
[ColumnInfo(name='dt_notific', description='Data da notificação', ...)]
>>> search_columns("sinan", endpoint="peste")
[ColumnInfo(name='con_classi', description='', ...)]

Portuguese-to-English column name mappings for SUS health datasets.

Each entry maps an uppercase Portuguese column name (as found in DATASUS files) to a lowercase_snake_case English equivalent. The mapping is curated from SINAN, SINASC, SIM, SIH, SIA, PNI, CNES, and CIHA documentation.

Examples

>>> from pysus.api.mappings import PT_TO_EN
>>> PT_TO_EN.get("DT_NOTIFIC")
'notification_date'
>>> PT_TO_EN.get("SG_UF")
'state'

pysus.api.mappings.to_english(df: DataFrame, store_mapping: bool = True) → DataFrame

Rename Portuguese column names to English equivalents.

Parameters:

• df (DataFrame) – Input DataFrame with Portuguese column names.

• store_mapping (bool, optional) – When True (default), stores the original-to-English mapping in df.attrs["aliases"] so it can be reversed later.

Returns:

A copy with English column names.

Return type:

DataFrame

Examples

>>> import pysus
>>> df = pysus.to_english(df)
>>> df.attrs["aliases"]  # original → English mapping
{'DT_NOTIFIC': 'notification_date', ...}

Errors & Warnings

PySUS exception hierarchy.

Every error carries a hint (how to fix it) and a docs_url (link to relevant documentation). The str method renders a user-friendly box with the error name, message, hint, and docs link.

exception pysus.api.errors.AuthenticationError(message: str = '', *, hint: str = '', docs_url: str = '')

Bases: PySUSError

Authentication or authorization failure (missing/wrong credentials).

hint: str = 'Set DADOSGOV_TOKEN env var or pass token= to the client.'

exception pysus.api.errors.CatalogError(message: str = '', *, hint: str = '', docs_url: str = '')

Bases: PySUSError

DuckLake catalog database operation failure (connect, query, schema).

hint: str = 'Ensure the DuckLake catalog is initialized (pysus ducklake init).'

exception pysus.api.errors.ConnectionError(message: str = '', *, hint: str = '', docs_url: str = '')

Bases: PySUSError

Failed to connect to a data source (DuckLake, FTP, DadosGov).

hint: str = 'Check your network connection and verify the data source is available.'

exception pysus.api.errors.ConversionError(message: str = '', *, hint: str = '', docs_url: str = '')

Bases: PySUSError

Error converting between file formats (e.g. DBF -> Parquet).

hint: str = 'The source file may be corrupted or in an unsupported format.'

exception pysus.api.errors.DownloadError(message: str = '', *, hint: str = '', docs_url: str = '')

Bases: PySUSError

File download failure (HTTP, S3, or FTP transfer error).

hint: str = 'Check your network connection and verify the file exists.'

exception pysus.api.errors.FormatError(message: str = '', *, hint: str = '', docs_url: str = '')

Bases: PySUSError

File format not recognized or not supported.

hint: str = 'Ensure the file is a supported format (Parquet, DBF, CSV, etc.).'

exception pysus.api.errors.ParseError(message: str = '', *, hint: str = '', docs_url: str = '')

Bases: PySUSError

Error parsing file contents or file metadata.

hint: str = 'The file may be corrupted. Try re-downloading.'

exception pysus.api.errors.PySUSError(message: str = '', *, hint: str = '', docs_url: str = '')

Bases: Exception

Base exception for all PySUS errors.

docs_url: str = 'https://pysus.readthedocs.io/en/latest/errors'

hint: str = ''

exception pysus.api.errors.PySUSWarning

Bases: UserWarning

Non-fatal warning for degraded operation.

exception pysus.api.errors.ValidationError(message: str = '', *, hint: str = '', docs_url: str = '')

Bases: PySUSError

Input validation error (missing fields, invalid values).

hint: str = 'Check the function signature and required parameters.'

pysus.api.errors.warn(message: str, *, stacklevel: int = 2) → None

Emit a PySUSWarning.

Cache Management

Smart cache utilities for PySUS.

Provides cache management: status, freshness check, and sync.

Usage:

from pysus.api.cache_utils import cache_status, is_cache_fresh

status = cache_status()
print(status["total_size_mb"])

class pysus.api.cache_utils.CacheStatus(path: Path, total_files: int, total_size_bytes: int, parquet_files: int, partial_files: int, last_modified: datetime | None)

Status of the PySUS cache directory.

last_modified: datetime | None

parquet_files: int

partial_files: int

path: Path

total_files: int

total_size_bytes: int

property total_size_mb: float

Total cache size in megabytes.

pysus.api.cache_utils.cache_status(cache_path: Path | None = None) → CacheStatus

Get status of the cache directory.

Parameters:

cache_path (Path, optional) – Cache directory. If None, uses default.

Returns:

Cache statistics.

Return type:

CacheStatus

pysus.api.cache_utils.clear_cache(cache_path: Path | None = None) → int

Clear all cached files.

Parameters:

cache_path (Path, optional) – Cache directory. If None, uses default.

Returns:

Number of files removed.

Return type:

int

pysus.api.cache_utils.format_size(size_bytes: int) → str

Format bytes as human-readable string.

Parameters:

size_bytes (int) – Size in bytes.

Returns:

Formatted string (e.g. ‘1.5 GB’).

Return type:

str

pysus.api.cache_utils.is_cache_fresh(local_path: Path, remote_mtime: datetime | None = None) → bool

Check if a cached file is still fresh.

Parameters:

• local_path (Path) – Path to the local cached file.

• remote_mtime (datetime, optional) – Last modification time of the remote file.

Returns:

True if the local file is up to date.

Return type:

bool

Input Validation

Input validation with fuzzy suggestions.

Uses difflib.get_close_matches to suggest corrections for typos.

Usage:

from pysus.api.validate import validate_choice

validate_choice("sinam", ["SINAN", "SIA", "SIM"])
# raises ValidationError with suggestion: Did you mean 'SINAN'?

pysus.api.validate.validate_choice(value: str, choices: list[str], label: str = 'value', cutoff: float = 0.6) → str

Validate that value is in choices, suggesting corrections.

Parameters:

• value (str) – User-provided value.

• choices (list[str]) – Allowed values.

• label (str) – Human-readable name for the parameter (for the error message).

• cutoff (float) – Similarity threshold for suggestions (0-1).

Returns:

The matched choice (case-insensitive match returned as canonical).

Return type:

str

Raises:

ValidationError – If value is not in choices and no close match is found.

pysus.api.validate.validate_dataset(name: str) → str

Validate a PySUS dataset name.

Parameters:

name (str) – Dataset name (e.g. ‘sinan’, ‘SIH’).

Returns:

Canonical dataset name.

Return type:

str

Raises:

ValidationError – If dataset name is not recognised.

pysus.api.validate.validate_origin(origin: str) → str

Validate a data origin.

Parameters:

origin (str) – Origin name (e.g. ‘ftp’, ‘dadosgov’).

Returns:

Canonical origin name.

Return type:

str

Raises:

ValidationError – If origin is not recognised.

pysus.api.validate.validate_source(source: str) → str

Validate a fetch source.

A source is where data is read from. "catalog" (default) serves the DuckLake catalog/S3 mirror; "origin" hits the origin server directly.

Parameters:

source (str) – Source name ("catalog" or "origin").

Returns:

Canonical source name.

Return type:

str

Raises:

ValidationError – If source is not recognised.

Data Quality

Column-level statistics for DataFrames.

Provides detailed per-column statistics including dtype, null counts, unique values, and memory usage.

Usage:

from pysus.api.quality.statistics import column_stats

stats = column_stats(df)
print(stats[["column", "null_pct", "memory_mb"]])

pysus.api.quality.statistics.column_stats(df: DataFrame) → DataFrame

Get detailed statistics for each column.

Parameters:

df (pd.DataFrame) – Input DataFrame.

Returns:

DataFrame with columns: column, dtype, null_count, null_pct, unique_count, unique_pct, memory_bytes, memory_mb, sample_value. Sorted by memory usage (descending).

Return type:

pd.DataFrame

Missing value analysis for DataFrames.

Provides per-column and per-state missing value percentages with configurable thresholds.

Usage:

from pysus.api.quality.missing import missing_values

result = missing_values(df, threshold=0.1)
grouped = missing_values(df, group_by="UF")

pysus.api.quality.missing.missing_values(df: DataFrame, group_by: str | None = None, threshold: float = 0.0) → DataFrame | tuple[DataFrame, DataFrame]

Analyze missing values in a DataFrame.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• group_by (str, optional) – Column to group by (e.g. "UF" for state-level analysis).

• threshold (float) – Minimum missing percentage to include (0.0-1.0).

Returns:

DataFrame with columns: column, missing_count, missing_pct, complete_count, complete_pct. If group_by is provided, returns a tuple of (summary_df, grouped_df).

Return type:

pd.DataFrame or tuple

Quality scoring for DataFrames.

Computes a 0-100 quality score based on completeness, validity, and consistency metrics.

Usage:

from pysus.api.quality.score import quality_score

score = quality_score(df)
print(f"Overall: {score.overall:.1f}/100")

class pysus.api.quality.score.QualityScore(overall: float, completeness: float, validity: float, consistency: float, details: dict[str, Any])

Bases: object

Quality score result.

overall

Overall score 0-100.

Type:

float

completeness

Completeness score 0-100 (non-null percentage).

Type:

float

validity

Validity score 0-100 (values passing basic checks).

Type:

float

consistency

Consistency score 0-100 (pattern matching).

Type:

float

details

Detailed breakdown by column.

Type:

dict

completeness: float

consistency: float

details: dict[str, Any]

overall: float

validity: float

pysus.api.quality.score.quality_score(df: DataFrame, schema: dict[str, dict[str, Any]] | None = None) → QualityScore

Calculate overall quality score for a DataFrame.

Scoring weights: - Completeness: 40% (average non-null percentage) - Validity: 40% (values passing basic validation) - Consistency: 20% (values matching expected patterns)

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• schema (dict, optional) – Column metadata for validation (from load_column_metadata).

Returns:

Quality score with overall and component scores.

Return type:

QualityScore

Basic data validation for DATASUS DataFrames.

Validates age ranges, date formats, categorical values, and user-defined rules.

Usage:

from pysus.api.quality.validation import validate_data

results = validate_data(df)
for r in results:
    if not r.passed:
        print(f"{r.column}: {r.details}")

class pysus.api.quality.validation.ValidationResult(column: str, rule: str, passed: bool, failed_count: int, failed_pct: float, details: str)

Bases: object

Result of a single validation rule.

column

Column name that was validated.

Type:

str

rule

Rule name (e.g. "age_range", "date_format").

Type:

str

passed

Whether validation passed.

Type:

bool

failed_count

Number of rows that failed.

Type:

int

failed_pct

Percentage of rows that failed (0.0-1.0).

Type:

float

details

Human-readable description of the result.

Type:

str

column: str

details: str

failed_count: int

failed_pct: float

passed: bool

rule: str

pysus.api.quality.validation.validate_data(df: DataFrame, rules: dict[str, dict[str, Any]] | None = None) → list[ValidationResult]

Validate DataFrame against built-in and custom rules.

Built-in rules: - Age columns (IDADE, NU_IDADE_N): 0-120 - Date columns (DT_*): YYYYMMDD format, reasonable range - Categorical columns (CS_*): values in expected set

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• rules (dict, optional) – Custom validation rules per column. Keys are column names, values are dicts with "type" and parameters.

Returns:

Validation results (includes passing rules).

Return type:

list[ValidationResult]

Automatic data profiling report for DataFrames.

Generates comprehensive profiling reports including statistics, missing values, quality scores, and distributions.

Usage:

from pysus.api.quality.profiling import profile_report

# Text report
print(profile_report(df))

# JSON report
report = profile_report(df, format="json")

# Save to file
profile_report(df, output="report.txt")

pysus.api.quality.profiling.profile_report(df: DataFrame, output: str | Path | None = None, format: Literal['html', 'json', 'text'] = 'text') → str | dict[str, Any]

Generate comprehensive profiling report.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• output (str or Path, optional) – File path to save report.

• format (str) – Output format: "html", "json", or "text".

Returns:

Report as string (text/html) or dict (json).

Return type:

str or dict

Data Transformation

Aggregation helpers for DATASUS DataFrames.

Provides state-level, age-group, and time-period aggregations.

Usage:

from pysus.api.transform.aggregation import aggregate_by_state

state_counts = aggregate_by_state(df, "DT_NOTIFIC")

pysus.api.transform.aggregation.aggregate_by_age_group(df: DataFrame, age_col: str, value_col: str, bins: list[int] | None = None) → DataFrame

Aggregate data by age groups.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• age_col (str) – Age column name.

• value_col (str) – Column to aggregate.

• bins (list, optional) – Age group boundaries. Default: [0, 5, 15, 30, 50, 70, 120].

Returns:

Age group aggregation.

Return type:

pd.DataFrame

pysus.api.transform.aggregation.aggregate_by_period(df: DataFrame, date_col: str, value_col: str, freq: Literal['M', 'Q', 'Y'] = 'M') → DataFrame

Aggregate data by time period.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• date_col (str) – Date column (YYYYMMDD format).

• value_col (str) – Column to aggregate.

• freq (str) – Frequency (``”M”``=monthly, ``”Q”``=quarterly, ``”Y”``=yearly).

Returns:

Period aggregation.

Return type:

pd.DataFrame

pysus.api.transform.aggregation.aggregate_by_state(df: DataFrame, value_col: str, agg_func: Literal['count', 'sum', 'mean', 'median'] = 'count') → DataFrame

Aggregate data by state (UF).

Parameters:

• df (pd.DataFrame) – Input DataFrame with UF column.

• value_col (str) – Column to aggregate.

• agg_func (str) – Aggregation function.

Returns:

State-level aggregation.

Return type:

pd.DataFrame

Column rename/alias system for historical DATASUS columns.

Maps old column names to current names for backward compatibility.

Usage:

from pysus.api.transform.aliases import get_aliases, rename_columns

old_names = get_aliases("sinan", "DT_NOTIFIC")
df = rename_columns(df, database="sinan")

pysus.api.transform.aliases.get_aliases(database: str, column: str) → list[str]

Get historical aliases for a column name.

Parameters:

• database (str) – Database name ("sinan", "sih", etc.).

• column (str) – Current column name.

Returns:

Previous names for this column.

Return type:

list[str]

pysus.api.transform.aliases.rename_columns(df: DataFrame, mapping: dict[str, str] | None = None, database: str | None = None) → DataFrame

Rename columns using mapping or historical aliases.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• mapping (dict, optional) – Custom rename mapping {old_name: new_name}.

• database (str, optional) – Database name to use historical aliases.

Returns:

DataFrame with renamed columns.

Return type:

pd.DataFrame

Cross-dataset linking helpers for DATASUS databases.

Provides common linking keys and merge utilities for joining datasets from different databases.

Usage:

from pysus.api.transform.linking import get_linking_keys, link_datasets

keys = get_linking_keys("sinan", "sih")
merged = link_datasets(sinan_df, sih_df, on="MUNIC_RES")

pysus.api.transform.linking.get_linking_keys(database1: str, database2: str) → list[str]

Get common linking keys between two databases.

Parameters:

• database1 (str) – First database name.

• database2 (str) – Second database name.

Returns:

Column names that can link the two datasets.

Return type:

list[str]

pysus.api.transform.linking.link_datasets(df1: DataFrame, df2: DataFrame, on: str | list[str], how: Literal['inner', 'left', 'right', 'outer'] = 'inner') → DataFrame

Link two DATASUS datasets on common keys.

Handles column name conflicts by adding suffixes.

Parameters:

• df1 (pd.DataFrame) – First DataFrame.

• df2 (pd.DataFrame) – Second DataFrame.

• on (str or list) – Column name(s) to join on.

• how (str) – Join type.

Returns:

Merged DataFrame.

Return type:

pd.DataFrame

Data masking for sensitive columns.

Provides reversible encryption, hashing, and redaction for sensitive data like CPF, names, and addresses.

Usage:

from pysus.api.transform.masking import mask_data, unmask_data

masked_df, key = mask_data(df, columns=["CPF", "NOME"])
original_df = unmask_data(masked_df, ["CPF", "NOME"], key)

pysus.api.transform.masking.mask_data(df: DataFrame, columns: list[str] | None = None, key: bytes | None = None, method: Literal['encrypt', 'hash', 'redact'] = 'encrypt') → tuple[DataFrame, bytes]

Mask sensitive data in DataFrame.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• columns (list, optional) – Columns to mask (auto-detect if None).

• key (bytes, optional) – Encryption key (generated if None).

• method (str) – Masking method.

Returns:

Masked DataFrame and key for reversal.

Return type:

tuple[pd.DataFrame, bytes]

pysus.api.transform.masking.unmask_data(df: DataFrame, columns: list[str], key: bytes) → DataFrame

Reverse encryption masking.

Parameters:

• df (pd.DataFrame) – DataFrame with masked columns.

• columns (list) – Columns to unmask.

• key (bytes) – Encryption key used for masking.

Returns:

Unmasked DataFrame.

Return type:

pd.DataFrame

Configurable numeric precision for DataFrames.

Controls float16/32/64 precision and memory optimization.

Usage:

from pysus.api.transform.precision import set_precision, optimize_memory

df32 = set_precision(df, precision="float32")
optimized = optimize_memory(df)

pysus.api.transform.precision.optimize_memory(df: DataFrame, downcast_int: bool = True, downcast_float: bool = True) → DataFrame

Optimize memory usage by downcasting numeric types.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• downcast_int (bool) – Downcast integer columns.

• downcast_float (bool) – Downcast float columns.

Returns:

Memory-optimized DataFrame.

Return type:

pd.DataFrame

pysus.api.transform.precision.set_precision(df: DataFrame, precision: Literal['float16', 'float32', 'float64'] = 'float32', columns: list[str] | None = None) → DataFrame

Set numeric precision for DataFrame columns.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• precision (str) – Target precision.

• columns (list, optional) – Specific columns (all numeric if None).

Returns:

DataFrame with adjusted precision.

Return type:

pd.DataFrame

Large dataset streaming for Parquet files.

Provides chunked iteration to process large files without loading them entirely into memory.

Usage:

from pysus.api.transform.streaming import stream_parquet

for chunk in stream_parquet("data.parquet", chunk_size=10000):
    process(chunk)

pysus.api.transform.streaming.stream_parquet(path: str | Path, chunk_size: int = 10000, columns: list[str] | None = None) → Generator[DataFrame, None, None]

Stream large Parquet files in chunks.

Parameters:

• path (str or Path) – Path to Parquet file.

• chunk_size (int) – Number of rows per chunk.

• columns (list, optional) – Columns to read.

Yields:

pd.DataFrame – DataFrame chunks.

Unit-of-measure detection for numeric columns.

Uses column names and value ranges to infer measurement units.

Usage:

from pysus.api.transform.units import detect_units

units = detect_units(df)
for u in units:
    print(f"{u.column}: {u.unit} ({u.confidence:.0%})")

class pysus.api.transform.units.DetectedUnit(column: str, unit: str, confidence: float, detected_from: Literal['column_name', 'value_range', 'metadata'])

Detected unit for a column.

column

Column name.

Type:

str

unit

Detected unit (e.g. "kg", "mg/dL").

Type:

str

confidence

Detection confidence 0.0-1.0.

Type:

float

detected_from

Detection method used.

Type:

str

column: str

confidence: float

detected_from: Literal['column_name', 'value_range', 'metadata']

unit: str

pysus.api.transform.units.detect_units(df: DataFrame, metadata: dict[str, dict[str, Any]] | None = None) → list[DetectedUnit]

Detect units of measurement for numeric columns.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• metadata (dict, optional) – Column metadata with "unit" keys.

Returns:

Detected units per column.

Return type:

list[DetectedUnit]

Export

Export utilities for PySUS DataFrames.

Provides CSV/Excel, GeoJSON, SQL DDL, and auto-detecting export().

Usage:

from pysus.api.export import export, to_csv, to_excel, to_geojson, to_sql

export(df, "output.csv")
export(df, "output.parquet")
export(df, "output.xlsx", sheet_name="Data")
to_geojson(df, "output.geojson", lat_col="LAT", lon_col="LON")
ddl = to_sql(df, "table_name", dialect="duckdb")

pysus.api.export.export(df: DataFrame, path: str | Path, **kwargs: object) → Path

Export DataFrame to a file, auto-detecting format from extension.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• path (str or Path) – Output file path.

• **kwargs – Additional arguments passed to the format-specific writer.

Returns:

Path to created file.

Return type:

Path

Raises:

• ValueError – If the file extension is not supported.

• Supported Formats –

• ----------------- –

• - .csv — CSV with metadata sidecar –

• - .xlsx — Excel with metadata sheet –

• - .parquet — Parquet (native pandas) –

• - .json — JSON lines –

• - .sqlite — SQLite table (requires table_name kwarg) –

pysus.api.export.to_csv(df: DataFrame, path: str | Path, metadata: dict | None = None, encoding: str = 'utf-8') → Path

Export DataFrame to CSV with optional metadata sidecar.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• path (str or Path) – Output file path.

• metadata (dict, optional) – Metadata to save in .metadata.json sidecar file.

• encoding (str) – File encoding.

Returns:

Path to created file.

Return type:

Path

pysus.api.export.to_excel(df: DataFrame, path: str | Path, metadata: dict | None = None, sheet_name: str = 'Data') → Path

Export DataFrame to Excel with metadata sheet.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• path (str or Path) – Output file path.

• metadata (dict, optional) – Metadata to include as a separate sheet.

• sheet_name (str) – Name of the data sheet.

Returns:

Path to created file.

Return type:

Path

pysus.api.export.to_geojson(df: DataFrame, path: str | Path, lat_col: str = 'LATITUDE', lon_col: str = 'LONGITUDE', geocode_col: str | None = None, properties: list[str] | None = None) → Path

Export DataFrame to GeoJSON with Point geometries.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• path (str or Path) – Output file path.

• lat_col (str) – Latitude column name.

• lon_col (str) – Longitude column name.

• geocode_col (str, optional) – Geocode column for properties.

• properties (list, optional) – Additional columns to include as properties.

Returns:

Path to created file.

Return type:

Path

pysus.api.export.to_sql(df: DataFrame, table_name: str, dialect: Literal['duckdb', 'mysql', 'postgresql', 'sqlite'] = 'duckdb', include_data: bool = False) → str

Generate SQL CREATE TABLE statement from DataFrame.

Parameters:

• df (pd.DataFrame) – Input DataFrame (used for schema inference).

• table_name (str) – Name for the SQL table.

• dialect (str) – SQL dialect.

• include_data (bool) – Include INSERT statements.

Returns:

SQL DDL string.

Return type:

str

CSV and Excel export with optional metadata.

Usage:

from pysus.api.export.csv_excel import to_csv, to_excel

to_csv(df, "output.csv", metadata={"source": "DATASUS"})
to_excel(df, "output.xlsx", metadata={"source": "DATASUS"})

pysus.api.export.csv_excel.to_csv(df: DataFrame, path: str | Path, metadata: dict | None = None, encoding: str = 'utf-8') → Path

Export DataFrame to CSV with optional metadata sidecar.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• path (str or Path) – Output file path.

• metadata (dict, optional) – Metadata to save in .metadata.json sidecar file.

• encoding (str) – File encoding.

Returns:

Path to created file.

Return type:

Path

pysus.api.export.csv_excel.to_excel(df: DataFrame, path: str | Path, metadata: dict | None = None, sheet_name: str = 'Data') → Path

Export DataFrame to Excel with metadata sheet.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• path (str or Path) – Output file path.

• metadata (dict, optional) – Metadata to include as a separate sheet.

• sheet_name (str) – Name of the data sheet.

Returns:

Path to created file.

Return type:

Path

GeoJSON export for geographic columns.

Exports DataFrames with lat/lon columns to GeoJSON FeatureCollection.

Usage:

from pysus.api.export.geojson import to_geojson

to_geojson(df, "output.geojson", lat_col="LATITUDE", lon_col="LONGITUDE")

pysus.api.export.geojson.to_geojson(df: DataFrame, path: str | Path, lat_col: str = 'LATITUDE', lon_col: str = 'LONGITUDE', geocode_col: str | None = None, properties: list[str] | None = None) → Path

Export DataFrame to GeoJSON with Point geometries.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• path (str or Path) – Output file path.

• lat_col (str) – Latitude column name.

• lon_col (str) – Longitude column name.

• geocode_col (str, optional) – Geocode column for properties.

• properties (list, optional) – Additional columns to include as properties.

Returns:

Path to created file.

Return type:

Path

SQL DDL export for DataFrames.

Generates CREATE TABLE statements for DuckDB (default), MySQL, PostgreSQL, and SQLite.

Usage:

from pysus.api.export.sql import to_sql

ddl = to_sql(df, "my_table", dialect="duckdb")
print(ddl)

pysus.api.export.sql.to_sql(df: DataFrame, table_name: str, dialect: Literal['duckdb', 'mysql', 'postgresql', 'sqlite'] = 'duckdb', include_data: bool = False) → str

Generate SQL CREATE TABLE statement from DataFrame.

Parameters:

• df (pd.DataFrame) – Input DataFrame (used for schema inference).

• table_name (str) – Name for the SQL table.

• dialect (str) – SQL dialect.

• include_data (bool) – Include INSERT statements.

Returns:

SQL DDL string.

Return type:

str

Data Diff

DataFrame comparison engine for version diffs.

Usage:

from pysus.api.diff.comparison import diff_dfs, diff_summary, diff_rows

result = diff_dfs(df_old, df_new, key_cols=["UF", "MUNIC_RES"])
summary = diff_summary(df_old, df_new)

class pysus.api.diff.comparison.DiffResult(columns_added: list[str] = <factory>, columns_removed: list[str] = <factory>, columns_type_changed: list[tuple[str, str, str]] = <factory>, rows_added: int = 0, rows_removed: int = 0, rows_modified: int = 0, rows_unchanged: int = 0, modified_cells: list[tuple[str, int, str, str]] = <factory>)

Bases: object

Result of a DataFrame comparison.

columns_added: list[str]

columns_removed: list[str]

columns_type_changed: list[tuple[str, str, str]]

modified_cells: list[tuple[str, int, str, str]]

rows_added: int = 0

rows_modified: int = 0

rows_removed: int = 0

rows_unchanged: int = 0

property summary: dict

Summary dict of the diff.

pysus.api.diff.comparison.diff_dfs(df_old: DataFrame, df_new: DataFrame, key_cols: list[str] | None = None) → DiffResult

Compare two DataFrames and return detailed diff.

Parameters:

• df_old (pd.DataFrame) – Older/previous version.

• df_new (pd.DataFrame) – Newer/current version.

• key_cols (list, optional) – Columns identifying rows for matching. If None, compares by position.

Returns:

Detailed diff result.

Return type:

DiffResult

pysus.api.diff.comparison.diff_rows(df_old: DataFrame, df_new: DataFrame, key_cols: list[str] | None = None) → DataFrame

Return rows that differ between two DataFrames.

Parameters:

• df_old (pd.DataFrame) – Older version.

• df_new (pd.DataFrame) – Newer version.

• key_cols (list, optional) – Columns for matching rows.

Returns:

DataFrame with diff info columns.

Return type:

pd.DataFrame

pysus.api.diff.comparison.diff_summary(df_old: DataFrame, df_new: DataFrame) → dict

Quick summary diff without detailed cell-level comparison.

Parameters:

• df_old (pd.DataFrame) – Older version.

• df_new (pd.DataFrame) – Newer version.

Returns:

Summary of changes.

Return type:

dict

Metadata Models

Metadata data models: facets, bags and merge logic.

The metadata layer is origin-agnostic. Every client (FTP, DadosGov, DuckLake, Saude) extracts metadata from a different source, but the shape that flows through PySUS is one: a MetadataBag with eight typed facets (identity, description, temporal, spatial, provenance, structure, access, quality) plus a raw dict that preserves unmapped fields.

The legacy dataclasses (Dataset, DatasetGroup, FileMeta, File, Column) remain in this module for backwards compatibility; new code should use MetadataBag.

class pysus.api.metadata.models.AccessFacet(url: str = '', format: str = '', size_bytes: int = 0, download_strategy: str = '', requires_auth: bool = False, policy: str = '', mime_type: str = '')

How to reach the data.

download_strategy: str = ''

format: str = ''

classmethod from_dict(data: dict[str, Any]) → AccessFacet

mime_type: str = ''

policy: str = ''

requires_auth: bool = False

size_bytes: int = 0

to_dict() → dict[str, Any]

url: str = ''

class pysus.api.metadata.models.Column(name: str, description: str, dtype: Annotated[str, AfterValidator(func=_validate_column_type)])

A column definition: name, description and dtype.

description: str

dtype: Annotated[str, AfterValidator(func=_validate_column_type)]

classmethod from_dict(data: dict[str, str]) → Column

Rehydrate a column from a serialized dict.

classmethod from_schema(name: str, dtype: Annotated[str, AfterValidator(func=_validate_column_type)], description: str = '') → Column

Create a Column with a description provided from the database.

name: str

to_dict() → dict[str, str]

Serialize the column.

pysus.api.metadata.models.DESCRIPTIVE_PRECEDENCE = ('saude', 'dadosgov', 'ftp', 'ducklake')

Origin precedence for descriptive metadata (titles, descriptions). Saude is the curated source; DuckLake is a mirror.

class pysus.api.metadata.models.Dataset(name: str, long_name: str, description: str)

Legacy dataset metadata record.

description: str

long_name: str

name: str

class pysus.api.metadata.models.DatasetGroup(name: str, long_name: str, description: str)

Legacy group metadata record.

description: str

long_name: str

name: str

class pysus.api.metadata.models.DescriptionFacet(title: str = '', long_name: str = '', description: str = '', tags: list[str] = <factory>, themes: list[str] = <factory>)

Human-readable title, description, tags and themes.

description: str = ''

classmethod from_dict(data: dict[str, Any]) → DescriptionFacet

long_name: str = ''

tags: list[str]

themes: list[str]

title: str = ''

to_dict() → dict[str, Any]

class pysus.api.metadata.models.File(origin: ~typing.Annotated[str, ~pydantic.functional_validators.AfterValidator(func=~pysus.api.types._validate_origin)], dataset: ~pysus.api.metadata.models.Dataset | None = None, group: ~pysus.api.metadata.models.DatasetGroup | None = None, columns: list[~pysus.api.metadata.models.Column] = <factory>, _meta: ~pysus.api.metadata.models.FileMeta | None = None)

Legacy file record.

columns: list[Column]

dataset: Dataset | None = None

group: DatasetGroup | None = None

origin: Annotated[str, AfterValidator(func=_validate_origin)]

class pysus.api.metadata.models.FileMeta(name: str, path: str, size: int, state: str | None = None, uf: str | None = None, year: int | None = None, month: int | None = None, origin_path: str | None = None, origin_size: int | None = None)

Legacy file metadata record.

month: int | None = None

name: str

origin_path: str | None = None

origin_size: int | None = None

path: str

size: int

state: str | None = None

uf: str | None = None

year: int | None = None

class pysus.api.metadata.models.IdentityFacet(name: str = '', slug: str = '', aliases: list[str] = <factory>, cross_origin_id: str = '')

Names and identifiers of the entity.

cross_origin_id is the shared CKAN UUID that links the same dataset across the dados.gov.br and dadosabertos.saude.gov.br portals.

aliases: list[str]

cross_origin_id: str = ''

classmethod from_dict(data: dict[str, Any]) → IdentityFacet

name: str = ''

slug: str = ''

to_dict() → dict[str, Any]

pysus.api.metadata.models.MODIFIED_PRECEDENCE = ('ducklake', 'ftp', 'dadosgov', 'saude')

Origin precedence for temporal.modified (upload time accuracy).

class pysus.api.metadata.models.MetadataBag(identity: ~pysus.api.metadata.models.IdentityFacet = <factory>, description: ~pysus.api.metadata.models.DescriptionFacet = <factory>, temporal: ~pysus.api.metadata.models.TemporalFacet = <factory>, spatial: ~pysus.api.metadata.models.SpatialFacet = <factory>, provenance: ~pysus.api.metadata.models.ProvenanceFacet = <factory>, structure: ~pysus.api.metadata.models.StructureFacet = <factory>, access: ~pysus.api.metadata.models.AccessFacet = <factory>, quality: ~pysus.api.metadata.models.QualityFacet = <factory>, raw: dict[str, ~typing.Any] = <factory>)

The canonical metadata container.

One bag per entity (client, dataset, group, file). Every facet has a documented merge rule (see merge_bags()) so bags from different origins can be combined without losing information.

access: AccessFacet

description: DescriptionFacet

classmethod from_dict(data: dict[str, Any]) → MetadataBag

Rehydrate a bag from a serialized dict.

identity: IdentityFacet

merge(other: MetadataBag, *, descriptive_precedence: tuple[str, ...] = ('saude', 'dadosgov', 'ftp', 'ducklake'), structure_precedence: tuple[str, ...] = ('ducklake', 'saude', 'dadosgov', 'ftp'), modified_precedence: tuple[str, ...] = ('ducklake', 'ftp', 'dadosgov', 'saude')) → MetadataBag

Return a new bag with other merged into self.

Per-facet rules (roadmap_saude.md §1.7):

• identity: first non-empty field; aliases union

• description: winner by descriptive_precedence; tags union

• temporal.created: earliest; temporal.modified: winner by modified_precedence; year/month first non-None

• spatial: most specific scope; ufs/municipalities union

• provenance: first non-empty per field; license most permissive; origins joined with /

• structure: columns winner by structure_precedence; row/file counts max

• access: url/size first non-zero; requires_auth OR

• quality: fingerprint first non-empty; integrity OR; freshness max

provenance: ProvenanceFacet

quality: QualityFacet

raw: dict[str, Any]

spatial: SpatialFacet

structure: StructureFacet

temporal: TemporalFacet

to_dict() → dict[str, Any]

Serialize the bag to JSON-compatible dicts.

class pysus.api.metadata.models.ProvenanceFacet(origin: str = '', organization: str = '', author: str = '', maintainer: str = '', contact: str = '', license: str = '', license_id: str = '', attribution: str = '', source_url: str = '')

Where the data came from and how it may be used.

attribution: str = ''

author: str = ''

contact: str = ''

classmethod from_dict(data: dict[str, Any]) → ProvenanceFacet

license: str = ''

license_id: str = ''

maintainer: str = ''

organization: str = ''

origin: str = ''

source_url: str = ''

to_dict() → dict[str, Any]

class pysus.api.metadata.models.QualityFacet(freshness_score: float | None = None, integrity_verified: bool = False, content_fingerprint: str = '', completeness_pct: float | None = None)

Integrity and completeness signals.

completeness_pct: float | None = None

content_fingerprint: str = ''

freshness_score: float | None = None

classmethod from_dict(data: dict[str, Any]) → QualityFacet

integrity_verified: bool = False

to_dict() → dict[str, Any]

pysus.api.metadata.models.SCOPE_RANK = {'local': 4, 'municipal': 3, 'national': 0, 'regional': 1, 'state': 2}

Geographic scope levels, from least to most specific. Used to merge spatial facets (never widen coverage).

pysus.api.metadata.models.STRUCTURE_PRECEDENCE = ('ducklake', 'saude', 'dadosgov', 'ftp')

Origin precedence for content/structure metadata. DuckLake holds the authoritative parquet schema and row counts.

class pysus.api.metadata.models.SpatialFacet(geographic_scope: str = 'national', ufs: list[str] = <factory>, municipalities: list[str] = <factory>, state: str | None = None)

Geographic coverage.

classmethod from_dict(data: dict[str, Any]) → SpatialFacet

geographic_scope: str = 'national'

municipalities: list[str]

state: str | None = None

to_dict() → dict[str, Any]

ufs: list[str]

class pysus.api.metadata.models.StructureFacet(columns: list[~pysus.api.metadata.models.Column] = <factory>, row_count: int = 0, file_count: int = 0, format: str = '', schema_fingerprint: str = '')

Schema and volume.

columns: list[Column]

file_count: int = 0

format: str = ''

classmethod from_dict(data: dict[str, Any]) → StructureFacet

row_count: int = 0

schema_fingerprint: str = ''

to_dict() → dict[str, Any]

class pysus.api.metadata.models.TemporalFacet(created: datetime | None = None, modified: datetime | None = None, periodicity: str | None = None, valid_from: datetime | None = None, valid_to: datetime | None = None, year: int | None = None, month: int | None = None)

Temporal coverage and freshness.

created: datetime | None = None

classmethod from_dict(data: dict[str, Any]) → TemporalFacet

modified: datetime | None = None

month: int | None = None

periodicity: str | None = None

to_dict() → dict[str, Any]

valid_from: datetime | None = None

valid_to: datetime | None = None

year: int | None = None

pysus.api.metadata.models.merge_bags(bags: list[MetadataBag], *, descriptive_precedence: tuple[str, ...] = ('saude', 'dadosgov', 'ftp', 'ducklake'), structure_precedence: tuple[str, ...] = ('ducklake', 'saude', 'dadosgov', 'ftp'), modified_precedence: tuple[str, ...] = ('ducklake', 'ftp', 'dadosgov', 'saude')) → MetadataBag

Merge several bags (typically one per origin) into one.

Bags whose provenance.origin is empty participate only through field-level fallbacks. All precedence tuples can be overridden.

class pysus.api.metadata.report.Columns

…

class pysus.api.metadata.report.Footer

…

class pysus.api.metadata.report.Header

…

OpenDataSUS Client

Async facade for the Saude (dadosabertos.saude.gov.br) client.

class pysus.api.saude.client.SaudeClient(*, cache_dir: Path | None = None, cache_ttl: timedelta = datetime.timedelta(days=1), timeout: float = 30.0, user_agent: str | None = None)

Bases: BaseRemoteClient

Async client for the OpenDataSUS portal.

The portal is a Next.js frontend over a CKAN backend. SaudeClient owns an httpx.AsyncClient and the on-disk caches for the Next.js buildId and the catalog pages. No authentication is required.

Example

>>> import asyncio
>>> from pysus.api.saude import SaudeClient
>>> async def main():
...     async with SaudeClient() as c:
...         datasets = await c.list_datasets(group="arboviroses")
...         print([d.name for d in datasets])
>>> asyncio.run(main())

BASE_URL: ClassVar[str] = 'https://dadosabertos.saude.gov.br'

property cache_dir: Path

Return the on-disk cache directory.

property cache_ttl: timedelta

Return the cache TTL.

async close() → None

Close the underlying HTTP client.

async connect() → None

Ensure the underlying HTTP client is ready (no-op).

async datasets(**kwargs) → list[SaudeDataset]

Return the pre-configured Saude theme datasets.

Each dataset is a SaudeDataset built from the static registry in pysus.api.saude.databases; content resolves the CKAN packages belonging to the theme.

property description: str

Return a description of the client.

async download(file: BaseRemoteFile, output: Path, callback: Callable[[int, int], None] | None = None) → Path

Download a remote Saude file to output.

async download_dataset(slug: str, *, dest_dir: Path | None = None, fmt: str | None = None, progress: Callable[[int, int], None] | None = None, overwrite: bool = False, use_cache: bool = True) → list[Path]

Download every downloadable resource of a dataset.

async download_resource(slug: str, *, resource_id: str | None = None, name: str | None = None, fmt: str | None = None, dest_dir: Path | None = None, progress: Callable[[int, int], None] | None = None, overwrite: bool = False, use_cache: bool = True) → Path

Download one resource of a dataset.

async fetch_dataset(slug: str, *, use_cache: bool = True) → CKANPackage

Fetch the full CKAN package for a single dataset.

async fetch_resources(slug: str, *, use_cache: bool = True) → list[Resource]

Fetch the resources of a dataset.

async iter_datasets(*, q: str | None = None, group: str | None = None, tag: str | None = None, fmt: str | None = None, max_pages: int | None = None, use_cache: bool = True) → AsyncIterator[CatalogEntry]

Yield every catalog entry across all pages.

async list_datasets(*, q: str | None = None, group: str | None = None, tag: str | None = None, fmt: str | None = None, page: int = 1, use_cache: bool = True) → list[CatalogEntry]

Return one page (20 entries) of the catalog listing.

async list_groups(*, use_cache: bool = True) → list[GroupRef]

Return the 14 catalog groups (themes).

async list_tags(*, use_cache: bool = True) → list[TagRef]

Return the catalog tags.

async login(**kwargs) → None

Authenticate — the portal is public, so this is a no-op.

property long_name: str

Return the human-readable client name.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the short client name.

YAML schema loader for Saude DEMAS endpoint columns.

Provides optional manual column descriptions that override the auto-inferred types from Parquet conversion. The YAML files live in pysus/api/saude/schemas/<dataset>.yaml and map endpoint names to lists of column definitions.

Usage:

from pysus.api.saude.schemas import load_endpoint_columns

cols = load_endpoint_columns("arboviroses", "dengue")
# [{"name": "nu_ano", "type": "integer",
#   "description_pt": "Ano da notificação", ...}, ...]

The loader is cached after the first read.

pysus.api.saude.schemas.apply_column_descriptions(columns_cursor: Any, dataset_id: int, dataset: str, endpoint: str) → int

Update column descriptions from YAML for endpoint.

Returns the number of columns updated.

pysus.api.saude.schemas.available_schemas() → list[str]

Return names of datasets that have YAML schema files.

pysus.api.saude.schemas.load_endpoint_columns(dataset: str, endpoint: str) → list[dict[str, Any]]

Return column definitions for endpoint within dataset.

Parameters:

• dataset (str) – Dataset name (e.g. "arboviroses").

• endpoint (str) – Endpoint name (e.g. "dengue").

Returns:

Each dict has keys name, type, description_pt, description_en. Returns an empty list when no schema file exists or the endpoint is not found.

Return type:

list[dict]

DuckLake Client

High-level client for DuckLake S3-based public health dataset catalog.

Provides authentication, dataset discovery, and file download capabilities backed by per-dataset DuckDB engines.

class pysus.api.ducklake.client.DuckLake(engine=None, columns_engine=None, update_on_close: bool = False, *, credentials: DuckLakeCredentials | None = None)

Bases: BaseRemoteClient

property catalog_adapter: CatalogAdapter

The central dataset-registry adapter (catalog.duckdb).

property catalog_path: Path

async close(update_catalog: bool | None = None) → None

Close the connection to the remote server.

property columns_adapter: ColumnsAdapter

The column-definitions adapter (catalog_columns.duckdb).

property columns_path: Path

async connect(force: bool = False, callback: Callable[[int, int], None] | None = None) → None

Establish a connection to the remote server.

credentials: DuckLakeCredentials | None

async datasets(**kwargs) → list[DuckDataset]

Return a list of available datasets matching kwargs.

property description: str

Return a textual description of the entity.

async download(file: BaseRemoteFile, output: Path, callback: Callable[[int, int], None] | None = None) → Path

Download a single file to output and return the local path.

async flush_catalogs(update: bool = True) → None

Upload dirty catalogs (if update) without re-downloading them.

Long-running writers use this to checkpoint: modified local databases are pushed to S3 while the local files stay attached. The fresh local copy just written is authoritative — there is no need to tear the adapters down and re-download the same catalog over the network (which was the source of previously observed hangs at checkpoint time).

get_dataset_adapter(name: str) → DatasetAdapter

Return (and register) the per-dataset adapter for name.

Creates a fresh adapter for datasets not yet seen by this client.

async login(**kwargs) → None

Authenticate with the remote server using kwargs credentials.

property long_name: str

Return the long / human-readable name.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the short name of the remote entity.

update_on_close: bool

Application-level models for DuckLake remote resources.

Wraps catalog ORM records into BaseRemoteFile, BaseRemoteDataset, and BaseRemoteGroup interfaces used by the rest of PySUS.

class pysus.api.ducklake.models.DuckDataset(*, client: DuckLake, group_definitions: dict[str, str] = {}, record: Dataset, border: Any, update_on_close: bool = False)

Bases: BaseRemoteDataset

property adapter: DatasetAdapter

border: Any

client: DuckLake

async close(update_catalog: bool | None = None)

async connect(force: bool = False) → None

property description: str

Return a textual description of the entity.

extractor_types: ClassVar[list] = [<class 'pysus.api.ducklake.metadata.DuckLakeDatasetExtractor'>]

Extractor classes the concrete subclass wants applied. Each client module (ftp/dadosgov/ducklake/saude) overrides this.

property id: int

property long_name: str

Return the long / human-readable name.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the short name of the remote entity.

async query(group: str | list[str] | None = None, state: str | list[str] | None = None, year: int | list[int] | range | None = None, month: int | list[int] | range | None = None) → list[File]

record: Dataset

async sync(update: bool = False) → None

update_on_close: bool

class pysus.api.ducklake.models.DuckGroup(*, dataset: BaseRemoteDataset)

Bases: BaseRemoteGroup

dataset: DuckDataset

property description: str

Return a textual description of the entity.

extractor_types: ClassVar[list] = [<class 'pysus.api.ducklake.metadata.DuckLakeGroupExtractor'>]

Extractor classes the concrete subclass wants applied. Each client module (ftp/dadosgov/ducklake/saude) overrides this.

property long_name: str

Return the long / human-readable name.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the short name of the remote entity.

record: Group

class pysus.api.ducklake.models.File(*, path: Path, type: str | Annotated[str, AfterValidator(func=_validate_file_type)], dataset: BaseRemoteDataset, group: BaseRemoteGroup | None = None)

Bases: BaseRemoteFile

property basename: str

Return the file name from the path.

property extension: str

Return the file extension string.

extractor_types: ClassVar[list] = [<class 'pysus.api.ducklake.metadata.DuckLakeFileExtractor'>]

Extractor classes the concrete subclass wants applied. Each client module (ftp/dadosgov/ducklake/saude) overrides this.

group: DuckGroup | None

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property modify: datetime

Return the last modification timestamp.

property record: File

property rows: int

property sha256: str | None

property size: int

Return the file size in bytes.

async verify(path: Path) → bool

FTP Client

Async FTP client wrapping the standard ftplib for DATASUS data access.

class pysus.api.ftp.client.FTP(*, host: str = 'ftp.datasus.gov.br', timeout: int = 60)

Bases: BaseRemoteClient

Async FTP client for navigating and downloading DATASUS data.

async close() → None

Close the FTP connection and reset the internal client state.

Raises:

Exception – Any exception raised by ftplib during disconnection.

async connect() → None

Establish the FTP connection to the remote host.

Raises:

Exception – Any exception raised by ftplib during connection.

async datasets(**kwargs) → list[Dataset]

Return a list of all available dataset instances for this client.

Returns:

A list of Dataset instances for all available databases.

Return type:

list[Dataset]

Raises:

ConnectionError – If the FTP client is not connected.

property description: str

Return a description of this client’s purpose.

Returns:

A description string explaining the FTP client’s capabilities.

Return type:

str

async download(file: BaseRemoteFile, output: Path, callback: Callable[[...], None] | None = None) → Path

Download a remote file locally, optionally reporting progress.

property ftp: FTP | None

Return the underlying ftplib.FTP, or None if not connected.

Returns:

The ftplib.FTP instance, or None if not connected.

Return type:

FTPLib | None

host: str

async login(**kwargs) → None

Authenticate and connect to the FTP server (alias for connect).

Parameters:

**kwargs – Forwarded to connect() (currently unused).

Raises:

Exception – Any exception raised by ftplib during authentication.

property long_name: str

Return the human-readable name of this client.

Returns:

The human-readable client name.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the short name of this client.

Returns:

The client short name (“FTP”).

Return type:

str

timeout: int

class pysus.api.ftp.client.FTPFileInfo

Bases: TypedDict

Parsed metadata for a file or directory entry from an FTP listing.

group: FTPGroupInfo | None

modify: datetime

month: int | None

name: str

size: int

state: State | None

type: str

year: int | None

class pysus.api.ftp.client.FTPGroupInfo

Bases: TypedDict

Metadata describing a file group within a dataset.

description: str | None

long_name: str | None

name: str

DATASUS FTP dataset definitions with filename parsers for each database.

class pysus.api.ftp.databases.CIHA(*, client: ~pysus.api.models.BaseRemoteClient, group_definitions: dict[str, str] = {'CIHA': 'Comunicação de Internação Hospitalar e Ambulatorial'}, paths: list[~pysus.api.ftp.models.Directory] = [<Directory: /dissemin/publicos/CIHA/201101_/Dados>])

Bases: Dataset

Comunicação de Internação Hospitalar e Ambulatorial (CIHA).

property description: str

Return a description of the dataset’s purpose.

Returns:

A description of the dataset’s purpose in Portuguese.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse a CIHA filename into group, state, year and month metadata.

Parameters:

filename (str) – The raw CIHA filename to parse.

Returns:

A dict with keys group, state, year, month. On parse failure values are set to None.

Return type:

dict[str, Any]

group_definitions: dict[str, str]

property long_name: str

Return the dataset full name in Portuguese.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the dataset short name.

Returns:

The dataset acronym (e.g. “CIHA”).

Return type:

str

paths: list[Directory]

class pysus.api.ftp.databases.CNES(*, client: ~pysus.api.models.BaseRemoteClient, group_definitions: dict[str, str] = {'DC': 'Dados Complementares', 'EE': 'Estabelecimento de Ensino', 'EF': 'Estabelecimento Filantrópico', 'EP': 'Equipes', 'EQ': 'Equipamentos', 'GM': 'Gestão e Metas', 'HB': 'Habilitação', 'IN': 'Incentivos', 'LT': 'Leitos', 'PF': 'Profissional', 'RC': 'Regra Contratual', 'SR': 'Serviço Especializado', 'ST': 'Estabelecimentos'}, paths: list[~pysus.api.ftp.models.Directory] = [<Directory: /dissemin/publicos/CNES/200508_/Dados>])

Bases: Dataset

Cadastro Nacional de Estabelecimentos de Saúde (CNES).

property description: str

Return a description of the dataset’s purpose.

Returns:

A description of the dataset’s purpose in Portuguese.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse a CNES filename into group, state, year and month metadata.

Parameters:

filename (str) – The raw CNES filename to parse.

Returns:

A dict with keys group, state, year, month. On parse failure values are set to None.

Return type:

dict[str, Any]

group_definitions: dict[str, str]

property long_name: str

Return the dataset full name in Portuguese.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the dataset short name.

Returns:

The dataset acronym (e.g. “CIHA”).

Return type:

str

paths: list[Directory]

class pysus.api.ftp.databases.IBGEDATASUS(*, client: ~pysus.api.models.BaseRemoteClient, group_definitions: dict[str, str] = {'ALF': 'Alfabetização', 'ESCA': 'Escolaridade A (Geral)', 'ESCB': 'Escolaridade B (Níveis Específicos)', 'IDOSO': 'População Idosa', 'POP': 'População', 'POPT': 'População Total', 'PROJ': 'Projeções Populacionais', 'RENDA': 'Rendimento e Renda'}, paths: list[~pysus.api.ftp.models.Directory] = [<Directory: /dissemin/publicos/IBGE/POP>, <Directory: /dissemin/publicos/IBGE/censo>, <Directory: /dissemin/publicos/IBGE/POPTCU>, <Directory: /dissemin/publicos/IBGE/projpop>])

Bases: Dataset

População Residente e Projeções (IBGE).

property description: str

Return a description of the dataset’s purpose.

Returns:

A description of the dataset’s purpose in Portuguese.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse an IBGE filename into group and year metadata.

Parameters:

filename (str) – The raw IBGE filename to parse.

Returns:

A dict with keys group, year. On parse failure values are set to None.

Return type:

dict[str, Any]

group_definitions: dict[str, str]

property long_name: str

Return the dataset full name in Portuguese.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the dataset short name.

Returns:

The dataset acronym (e.g. “CIHA”).

Return type:

str

paths: list[Directory]

class pysus.api.ftp.databases.PNI(*, client: ~pysus.api.models.BaseRemoteClient, group_definitions: dict[str, str] = {'CPNI': 'Cobertura Vacinal', 'DPNI': 'Doses Aplicadas'}, paths: list[~pysus.api.ftp.models.Directory] = [<Directory: /dissemin/publicos/PNI/DADOS>])

Bases: Dataset

Programa Nacional de Imunizações (PNI).

property description: str

Return a description of the dataset’s purpose.

Returns:

A description of the dataset’s purpose in Portuguese.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse a PNI filename into group, state and year metadata.

Parameters:

filename (str) – The raw PNI filename to parse.

Returns:

A dict with keys group, state, year. On parse failure values are set to None.

Return type:

dict[str, Any]

group_definitions: dict[str, str]

property long_name: str

Return the dataset full name in Portuguese.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the dataset short name.

Returns:

The dataset acronym (e.g. “CIHA”).

Return type:

str

paths: list[Directory]

class pysus.api.ftp.databases.SIA(*, client: ~pysus.api.models.BaseRemoteClient, group_definitions: dict[str, str] = {'AB': 'Atenção Básica', 'AC': 'Alta Complexidade', 'AD': 'APAC de Laudos Diversos', 'AM': 'APAC de Medicamentos', 'AN': 'APAC de Nefrologia', 'AQ': 'APAC de Quimioterapia', 'AR': 'APAC de Radioterapia', 'AT': 'Atenção', 'BI': 'Boletim de Produção Ambulatorial Individualizado', 'PA': 'Produção Ambulatorial', 'PS': 'Procedimentos Especiais', 'SA': 'Saúde'}, paths: list[~pysus.api.ftp.models.Directory] = [<Directory: /dissemin/publicos/SIASUS/199407_200712/Dados>, <Directory: /dissemin/publicos/SIASUS/200801_/Dados>])

Bases: Dataset

Sistema de Informações Ambulatoriais — outpatient information system.

property description: str

Return a description of the dataset’s purpose.

Returns:

A description of the dataset’s purpose in Portuguese.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse an SIA filename into group, state, year and month metadata.

Handles DATASUS part-split files: BIRJ2504_2.dbc is the second part of group BI, state RJ, 2025-04 — the _2 suffix must not leak into the month.

Parameters:

filename (str) – The raw SIA filename to parse.

Returns:

A dict with keys group, state, year, month. On parse failure values are set to None.

Return type:

dict[str, Any]

group_definitions: dict[str, str]

property long_name: str

Return the dataset full name in Portuguese.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the dataset short name.

Returns:

The dataset acronym (e.g. “CIHA”).

Return type:

str

paths: list[Directory]

class pysus.api.ftp.databases.SIH(*, client: ~pysus.api.models.BaseRemoteClient, group_definitions: dict[str, str] = {'ER': 'AIH Rejeitada com Erro', 'RD': 'AIH Reduzida', 'RJ': 'AIH Rejeitada', 'SP': 'Serviços Profissionais'}, paths: list[~pysus.api.ftp.models.Directory] = [<Directory: /dissemin/publicos/SIHSUS/199201_200712/Dados>, <Directory: /dissemin/publicos/SIHSUS/200801_/Dados>])

Bases: Dataset

Sistema de Informações Hospitalares (SIH).

property description: str

Return a description of the dataset’s purpose.

Returns:

A description of the dataset’s purpose in Portuguese.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse an SIH filename into group, state, year and month metadata.

Parameters:

filename (str) – The raw SIH filename to parse.

Returns:

A dict with keys group, state, year, month. On parse failure values are set to None.

Return type:

dict[str, Any]

group_definitions: dict[str, str]

property long_name: str

Return the dataset full name in Portuguese.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the dataset short name.

Returns:

The dataset acronym (e.g. “CIHA”).

Return type:

str

paths: list[Directory]

class pysus.api.ftp.databases.SIM(*, client: ~pysus.api.models.BaseRemoteClient, group_definitions: dict[str, str] = {'DO': 'Mortalidade Geral (CID-10)', 'DOR': 'Mortalidade Geral (CID-9)'}, paths: list[~pysus.api.ftp.models.Directory] = [<Directory: /dissemin/publicos/SIM/CID10/DORES>, <Directory: /dissemin/publicos/SIM/CID9/DORES>])

Bases: Dataset

Sistema de Informação sobre Mortalidade (SIM).

property description: str

Return a description of the dataset’s purpose.

Returns:

A description of the dataset’s purpose in Portuguese.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse a SIM filename into group, state and year metadata.

Parameters:

filename (str) – The raw SIM filename to parse.

Returns:

A dict with keys group, state, year. On parse failure values are set to None.

Return type:

dict[str, Any]

group_definitions: dict[str, str]

property long_name: str

Return the dataset full name in Portuguese.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the dataset short name.

Returns:

The dataset acronym (e.g. “CIHA”).

Return type:

str

paths: list[Directory]

class pysus.api.ftp.databases.SINAN(*, client: ~pysus.api.models.BaseRemoteClient, group_definitions: dict[str, str] = {'ACBI': 'Acidente de trabalho com material biológico', 'ACGR': 'Acidente de trabalho', 'ANIM': 'Acidente por Animais Peçonhentos', 'ANTR': 'Atendimento Antirrabico', 'BOTU': 'Botulismo', 'CANC': 'Cancêr relacionado ao trabalho', 'CHAG': 'Doença de Chagas Aguda', 'CHIK': 'Febre de Chikungunya', 'COLE': 'Cólera', 'COQU': 'Coqueluche', 'DENG': 'Dengue', 'DERM': 'Dermatoses ocupacionais', 'DIFT': 'Difteria', 'ESQU': 'Esquistossomose', 'EXAN': 'Doença exantemáticas', 'FMAC': 'Febre Maculosa', 'FTIF': 'Febre Tifóide', 'HANS': 'Hanseníase', 'HANT': 'Hantavirose', 'HEPA': 'Hepatites Virais', 'IEXO': 'Intoxicação Exógena', 'INFL': 'Influenza Pandêmica', 'LEIV': 'Leishmaniose Visceral', 'LEPT': 'Leptospirose', 'LERD': 'LER/Dort', 'LTAN': 'Leishmaniose Tegumentar Americana', 'MALA': 'Malária', 'MENI': 'Meningite', 'MENT': 'Transtornos mentais relacionados ao trabalho', 'NTRA': 'Notificação de Tracoma', 'PAIR': 'Perda auditiva por ruído relacionado ao trabalho', 'PEST': 'Peste', 'PFAN': 'Paralisia Flácida Aguda', 'PNEU': 'Pneumoconioses realacionadas ao trabalho', 'RAIV': 'Raiva', 'SDTA': 'Surto Doenças Transmitidas por Alimentos', 'SIFA': 'Sífilis Adquirida', 'SIFC': 'Sífilis Congênita', 'SIFG': 'Sífilis em Gestante', 'SRC': 'Síndrome da Rubéola Congênia', 'TETA': 'Tétano Acidental', 'TETN': 'Tétano Neonatal', 'TOXC': 'Toxoplasmose Congênita', 'TOXG': 'Toxoplasmose Gestacional', 'TRAC': 'Inquérito de Tracoma', 'TUBE': 'Tuberculose', 'VARC': 'Varicela', 'VIOL': 'Violência doméstica, sexual e/ou outras violências', 'ZIKA': 'Zika Vírus'}, paths: list[~pysus.api.ftp.models.Directory] = [<Directory: /dissemin/publicos/SINAN/DADOS/FINAIS>, <Directory: /dissemin/publicos/SINAN/DADOS/PRELIM>])

Bases: Dataset

Sistema de Informação de Agravos de Notificação (SINAN).

property description: str

Return a description of the dataset’s purpose.

Returns:

A description of the dataset’s purpose in Portuguese.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse a SINAN filename into group and year metadata.

Parameters:

filename (str) – The raw SINAN filename to parse.

Returns:

A dict with keys group, year. On parse failure values are set to None.

Return type:

dict[str, Any]

group_definitions: dict[str, str]

property long_name: str

Return the dataset full name in Portuguese.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the dataset short name.

Returns:

The dataset acronym (e.g. “CIHA”).

Return type:

str

paths: list[Directory]

class pysus.api.ftp.databases.SINASC(*, client: ~pysus.api.models.BaseRemoteClient, group_definitions: dict[str, str] = {'DN': 'Declarações de Nascidos Vivos', 'DNR': 'Nascidos Vivos por UF de residência'}, paths: list[~pysus.api.ftp.models.Directory] = [<Directory: /dissemin/publicos/SINASC/NOV/DNRES>, <Directory: /dissemin/publicos/SINASC/ANT/DNRES>])

Bases: Dataset

Sistema de Informações sobre Nascidos Vivos (SINASC).

property description: str

Return a description of the dataset’s purpose.

Returns:

A description of the dataset’s purpose in Portuguese.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse a SINASC filename into group, state and year metadata.

Parameters:

filename (str) – The raw SINASC filename to parse.

Returns:

A dict with keys group, state, year. On parse failure values are set to None.

Return type:

dict[str, Any]

group_definitions: dict[str, str]

property long_name: str

Return the dataset full name in Portuguese.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the dataset short name.

Returns:

The dataset acronym (e.g. “CIHA”).

Return type:

str

paths: list[Directory]

Data model classes for FTP directories, files, groups and datasets.

class pysus.api.ftp.models.Dataset(*, client: BaseRemoteClient, group_definitions: dict[str, str] = {}, paths: list[Directory] = [])

Bases: BaseRemoteDataset, ABC

Abstract base for a DATASUS dataset, providing file discovery via FTP.

abstract property description: str

Return a description of the dataset’s purpose.

Returns:

A description of the dataset’s purpose.

Return type:

str

extractor_types: ClassVar[list] = [<class 'pysus.api.ftp.metadata.FtpDatasetExtractor'>]

Extractor classes the concrete subclass wants applied. Each client module (ftp/dadosgov/ducklake/saude) overrides this.

abstractmethod formatter(filename: str) → dict[str, Any]

Parse a filename into metadata (group, state, year, etc.).

Parameters:

filename (str) – The raw filename to parse.

Returns:

A dictionary of parsed metadata fields.

Return type:

dict[str, Any]

group_definitions: dict[str, str]

abstract property long_name: str

Return the dataset full name in Portuguese.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

abstract property name: str

Return the dataset short name.

Returns:

The dataset acronym.

Return type:

str

paths: list[Directory]

class pysus.api.ftp.models.Directory(path: str, parent: Directory | Dataset | Group | None = None, client: BaseRemoteClient | None = None, formatter: Callable | None = None, dataset: Dataset | None = None)

Bases: object

A remote FTP directory lazily loaded into files and subdirectories.

property content: list[Directory | File]

Return the directory contents, loading from FTP if not yet cached.

Returns:

The list of files and subdirectories.

Return type:

list[Directory | File]

async load() → None

Fetch and parse the directory listing from the FTP server.

Raises:

ValueError – If the client is not an FTP instance.

class pysus.api.ftp.models.File(*, path: Path, type: str | Annotated[str, AfterValidator(func=_validate_file_type)], dataset: BaseRemoteDataset, group: BaseRemoteGroup | None = None)

Bases: BaseRemoteFile

A single file on the DATASUS FTP server with parsed metadata.

property extension: str

Return the file extension (e.g. .dbc, .dbf).

Returns:

The file extension including the leading dot.

Return type:

str

extractor_types: ClassVar[list] = [<class 'pysus.api.ftp.metadata.FtpFileExtractor'>]

Extractor classes the concrete subclass wants applied. Each client module (ftp/dadosgov/ducklake/saude) overrides this.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property modify: datetime

Return the last modification timestamp.

Returns:

The file’s last modification datetime.

Return type:

datetime

Raises:

ValueError – If no modification date is available.

property month: int | None

Return the data month extracted from the filename, if available.

Returns:

The month as an integer, or None if not available.

Return type:

int | None

property size: int

Return the file size in bytes.

Returns:

The file size in bytes.

Return type:

int

property state: Annotated[str, AfterValidator(func=_validate_state)] | None

Return the state code extracted from the filename, if available.

Returns:

The state code, or None if not available.

Return type:

State | None

property year: int | None

Return the data year extracted from the filename, if available.

Returns:

The year as an integer, or None if not available.

Return type:

int | None

class pysus.api.ftp.models.Group(name: str, path: str, dataset: Dataset, long_name: str, description: str = '')

Bases: BaseRemoteGroup

A group of related files within a dataset (e.g. all files of a type).

property content: list[Directory | File]

Return the contents of the underlying directory.

Returns:

The directory contents.

Return type:

list[Directory | File]

property description: str

Return the group description.

Returns:

The group description.

Return type:

str

extractor_types: ClassVar[list] = [<class 'pysus.api.ftp.metadata.FtpGroupExtractor'>]

Extractor classes the concrete subclass wants applied. Each client module (ftp/dadosgov/ducklake/saude) overrides this.

property long_name: str

Return the human-readable group name.

Returns:

The human-readable group name.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the group short code (e.g. ‘RD’, ‘PA’).

Returns:

The group short code.

Return type:

str

path: str

DadosGov Client

HTTP client and data models for the dados.gov.br API.

class pysus.api.dadosgov.client.ConjuntoDados(*, client: ~pysus.api.models.BaseRemoteClient | None = None, id: str, titulo: str, nome: str, recursos: list[~pysus.api.dadosgov.client.Recurso] = <factory>)

Bases: BaseModel

A dataset group as returned by the dados.gov.br API.

client: BaseRemoteClient | None

id: str

resources: list[Recurso]

slug: str

title: str

class pysus.api.dadosgov.client.DadosGov(*, base_url: str = 'https://dados.gov.br/dados/api')

Bases: BaseRemoteClient

Client for the dados.gov.br open data portal API.

base_url: str

async close() → None

Close the underlying HTTP client and release resources.

async connect(token: str | None = None) → None

Connect to the dados.gov.br API with the given token.

Parameters:

token (str, optional) – The API authentication token. If not provided, uses the previously stored token.

Raises:

ValueError – If no token is provided and none was previously stored.

async datasets(**kwargs) → list[Dataset]

Return a list of pre-configured health datasets.

Returns:

A list of available Dataset instances for known health databases.

Return type:

list[Dataset]

property description: str

Return a description of the client.

Returns:

A Portuguese description of the API interface.

Return type:

str

async download(file: BaseRemoteFile, output: Path, callback: Callable[[int, int], None] | None = None) → Path

Download a remote file to a local path.

async get_dataset(id: str) → ConjuntoDados

Fetch a single dataset by its ID.

Parameters:

id (str) – The unique identifier of the dataset.

Returns:

The requested dataset.

Return type:

ConjuntoDados

Raises:

ConnectionError – If the client is not connected.

async list_datasets(**kwargs) → list[ConjuntoDados]

Search and list available datasets from the portal.

Parameters:

**kwargs –

Search parameters. Supported keys:

• pagina (int): Page number for pagination.

• nome_conjunto (str): Filter by dataset name.

• dados_abertos (bool): Filter by open data flag.

• is_privado (bool): Filter by private datasets.

• id_organizacao (str): Filter by organisation ID.

Returns:

A list of datasets matching the search criteria.

Return type:

list[ConjuntoDados]

Raises:

ConnectionError – If the client is not connected.

async login(token: str | None = None, **kwargs) → None

Authenticate with the API.

Delegates to the connect() method.

Parameters:

• token (str, optional) – The API authentication token.

• **kwargs – Additional keyword arguments (currently unused).

property long_name: str

Return the human-readable client name.

Returns:

The full Portuguese name of the portal.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the short client name.

Returns:

The abbreviated client name "DadosGov".

Return type:

str

class pysus.api.dadosgov.client.Recurso(*, id: str, titulo: str, link: str, tamanho: int, dataUltimaAtualizacaoArquivo: Annotated[datetime | None, BeforeValidator(func=to_datetime, json_schema_input_type=PydanticUndefined)] = None, nomeArquivo: str | None = None)

Bases: BaseModel

A single resource (file) within a dataset on dados.gov.br.

api_size: int

file_name: str | None

async get_size() → int

Retrieve the file size from the remote server.

Makes a HEAD request, falling back to GET with a Range: bytes=0-0 header when the server rejects HEAD. The fallback answers with a 206 whose Content-Length is the length of the single requested byte (i.e. 1); the real total lives in Content-Range.

Returns:

The file size in bytes, or 0 if the size could not be determined.

Return type:

int

id: str

last_modified: DateTime

title: str

url: str

pysus.api.dadosgov.client.to_bool(value: Any) → bool

Parse a Brazilian Portuguese boolean value into a bool.

Parameters:

value (Any) – The value to parse (e.g., "sim", "não", True, False).

Returns:

True if the value represents an affirmative, False otherwise.

Return type:

bool

pysus.api.dadosgov.client.to_datetime(value: Any) → datetime | None

Parse a Brazilian date string into a datetime object.

Parameters:

value (Any) – The value to parse, expected to be a date string in Brazilian format (e.g., %d/%m/%Y %H:%M:%S or %d/%m/%Y).

Returns:

Parsed datetime object, or None if the value cannot be parsed.

Return type:

datetime or None

Pre-configured health database definitions accessible via dados.gov.br.

class pysus.api.dadosgov.databases.CNES(*, client: DadosGov, group_definitions: dict[str, str] = {}, ids: list[str] = ['40a0d093-b12f-44a4-bdc7-bae8eb54dd04', '9455b341-b06e-408e-8e10-54b32b3d74ec'], group_aliases: dict[str, str] = {})

Bases: Dataset

Cadastro Nacional de Estabelecimentos de Saúde (CNES).

property description: str

Return a description of the dataset.

Returns:

A Portuguese description of the CNES information system.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse a CNES filename and extract metadata.

Parameters:

filename (str) – The name of the file to parse.

Returns:

A dictionary with keys state, year, and month. Unrecognised files return None for all keys.

Return type:

dict[str, Any]

ids: list[str]

property long_name: str

Return the human-readable name.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the short name.

Returns:

The abbreviated dataset name "CNES".

Return type:

str

class pysus.api.dadosgov.databases.COVID19(*, client: DadosGov, group_definitions: dict[str, str] = {}, ids: list[str] = ['1ba1801e-aec0-4dba-ae2a-7732f0a0c9f7'], group_aliases: dict[str, str] = {})

Bases: Dataset

Casos Confirmados de COVID-19.

property description: str

Return a description of the dataset.

Returns:

A Portuguese description of the COVID-19 confirmed cases dataset.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse a COVID-19 filename and extract metadata.

Parameters:

filename (str) – The name of the file to parse.

Returns:

A dictionary with keys state, year, and month. Unrecognised files return None for all keys.

Return type:

dict[str, Any]

ids: list[str]

property long_name: str

Return the human-readable name.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the short name.

Returns:

The abbreviated dataset name "COVID19".

Return type:

str

class pysus.api.dadosgov.databases.PNI(*, client: DadosGov, group_definitions: dict[str, str] = {}, ids: list[str] = ['2989d396-cb09-47e7-a3b8-a4b951ca0200', '543aa08a-46c4-44e8-802e-198daa30753d', '04292d08-ee4f-463a-b7b5-76cfb76775b3', '7ed6eecc-c254-475c-92c5-daba5727596b', '783b7456-6a6c-4025-a8bd-8e9caa0fb962', 'c6c3c6f3-2026-48a2-84ac-d8039714a0ba', '9a25b796-80e3-444a-a4e7-405f5596d8ab'], group_aliases: dict[str, str] = {'dataset-doses-aplicadas-pelo-programa-de-nacional-de-imunizacoes-pni_2022': 'DPNI', 'doses-aplicadas-pelo-programa-de-nacional-de-imunizacoes-pni': 'DPNI', 'doses-aplicadas-pelo-programa-de-nacional-de-imunizacoes-pni-2020': 'DPNI', 'doses-aplicadas-pelo-programa-de-nacional-de-imunizacoes-pni-2021': 'DPNI', 'doses-aplicadas-pelo-programa-de-nacional-de-imunizacoes-pni-2023': 'DPNI', 'doses-aplicadas-pelo-programa-de-nacional-de-imunizacoes-pni-2025': 'DPNI', 'doses-aplicadas-pelo-programa-de-nacional-de-imunizacoes-pni-2026': 'DPNI'})

Bases: Dataset

Programa Nacional de Imunizações (PNI).

property description: str

Return a description of the dataset.

Returns:

A Portuguese description of the PNI vaccination monitoring system.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse a PNI vaccination filename into month and year.

Parameters:

filename (str) – The name of the file to parse.

Returns:

A dictionary with keys state, year, and month. Unrecognised files return None for all keys.

Return type:

dict[str, Any]

group_aliases: dict[str, str]

ids: list[str]

property long_name: str

Return the human-readable name.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the short name.

Returns:

The abbreviated dataset name "PNI".

Return type:

str

class pysus.api.dadosgov.databases.SIA(*, client: DadosGov, group_definitions: dict[str, str] = {}, ids: list[str] = ['9a335cb7-2b4f-4fce-8947-e8441b4a90af'], group_aliases: dict[str, str] = {})

Bases: Dataset

Sistema de Informações Ambulatoriais (SIA).

property description: str

Return a description of the dataset.

Returns:

A Portuguese description of the SIA outpatient information system.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse an SIA filename into year.

Parameters:

filename (str) – The name of the file to parse.

Returns:

A dictionary with keys state, year, and month. Unrecognised files return None for all keys.

Return type:

dict[str, Any]

ids: list[str]

property long_name: str

Return the human-readable name.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the short name.

Returns:

The abbreviated dataset name "SIA".

Return type:

str

class pysus.api.dadosgov.databases.SIM(*, client: DadosGov, group_definitions: dict[str, str] = {}, ids: list[str] = ['5f121f4d-47c6-428e-8ec6-e8ec56417172'], group_aliases: dict[str, str] = {'sim-1979-2019': 'DO'})

Bases: Dataset

Sistema de Informação sobre Mortalidade (SIM).

property description: str

Return a description of the dataset.

Returns:

A Portuguese description of the SIM mortality information system.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse a SIM filename into year.

Parameters:

filename (str) – The name of the file to parse.

Returns:

A dictionary with keys state, year, and month. Unrecognised files return None for all keys.

Return type:

dict[str, Any]

group_aliases: dict[str, str]

ids: list[str]

property long_name: str

Return the human-readable name.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the short name.

Returns:

The abbreviated dataset name "SIM".

Return type:

str

class pysus.api.dadosgov.databases.SINAN(*, client: DadosGov, group_definitions: dict[str, str] = {}, ids: list[str] = ['4d5e5d44-58a8-4d67-b8aa-4ef1e4b00a1c', '5699abe0-0510-4da8-b47d-209b3bb32b34', '4557ba96-7d52-4a56-bd6f-f99a5af09f77', '740ce8f4-7a5d-4351-aad4-7623f2490ada', 'cf044c1b-b966-4d0e-bab0-f3aa65897b7d', '2d4997fb-cd11-4ce2-b217-09cd50e3151f', '8a585222-4c2e-43b7-807d-59355ee79c48', '527e8665-de64-4f81-b7c3-40b59c7d1d3c'], group_aliases: dict[str, str] = {'arboviroses-dengue': 'DENG', 'arboviroses-febre-de-chikungunya': 'CHIK', 'arboviroses-zika-virus': 'ZIKA', 'dados-tuberculose': 'TUBE', 'hanseniase': 'HANS', 'sifilis': 'SIFA'})

Bases: Dataset

Sistema de Informação de Agravos de Notificação (SINAN).

property description: str

Return a description of the dataset.

Returns:

A Portuguese description of the SINAN notifiable diseases system.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse a SINAN filename into state and year.

Parameters:

filename (str) – The name of the file to parse.

Returns:

A dictionary with keys state, year, and month. Unrecognised files return None for all keys.

Return type:

dict[str, Any]

group_aliases: dict[str, str]

ids: list[str]

property long_name: str

Return the human-readable name.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the short name.

Returns:

The abbreviated dataset name "SINAN".

Return type:

str

class pysus.api.dadosgov.databases.SINASC(*, client: DadosGov, group_definitions: dict[str, str] = {}, ids: list[str] = ['441cc6bd-684a-4afd-a88b-ba4734c9e83e'], group_aliases: dict[str, str] = {'sistema-de-informacao-sobre-nascidos-vivos-sinasc-1996-a-20201': 'DN'})

Bases: Dataset

Sistema de Informações sobre Nascidos Vivos (SINASC).

property description: str

Return a description of the dataset.

Returns:

Portuguese description of the SINASC live birth system.

Return type:

str

formatter(filename: str) → dict[str, Any]

Parse a SINASC filename into year.

Parameters:

filename (str) – The name of the file to parse.

Returns:

A dictionary with keys state, year, and month. Unrecognised files return None for all keys.

Return type:

dict[str, Any]

group_aliases: dict[str, str]

ids: list[str]

property long_name: str

Return the human-readable name.

Returns:

The full Portuguese name of the dataset.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the short name.

Returns:

The abbreviated dataset name "SINASC".

Return type:

str

Internal domain models for datasets, groups, and files from dados.gov.br.

class pysus.api.dadosgov.models.Dataset(*, client: DadosGov, group_definitions: dict[str, str] = {}, ids: list[str] = [], group_aliases: dict[str, str] = {})

Bases: BaseRemoteDataset

A health dataset available through dados.gov.br.

Subclasses define a list of API dataset IDs and an optional formatter() that extracts metadata from file names.

client: DadosGov

extractor_types: ClassVar[list] = [<class 'pysus.api.dadosgov.metadata.DadosGovDatasetExtractor'>]

Extractor classes the concrete subclass wants applied. Each client module (ftp/dadosgov/ducklake/saude) overrides this.

abstractmethod formatter(filename: str) → dict[str, Any]

Extract structured metadata from a filename.

group_aliases: dict[str, str]

ids: list[str]

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

class pysus.api.dadosgov.models.File(*, path: Path, type: str = 'File', dataset: BaseRemoteDataset, group: BaseRemoteGroup | None = None, record: Recurso)

Bases: BaseRemoteFile

A downloadable file from a dados.gov.br dataset.

property extension: str

Return the file extension.

Returns:

The file extension (e.g., ".csv", ".zip").

Return type:

str

extractor_types: ClassVar[list] = [<class 'pysus.api.dadosgov.metadata.DadosGovFileExtractor'>]

Extractor classes the concrete subclass wants applied. Each client module (ftp/dadosgov/ducklake/saude) overrides this.

async fetch_metadata() → None

Fetch file size and last-modified from the remote server.

Updates record.api_size and record.last_modified in-place. Silently ignores connection errors.

async fetch_size() → int

Fetch the remote file size and update the local record.

Makes a HEAD request, falling back to GET with a Range: bytes=0-0 header when the server rejects HEAD. In the ranged response the total size comes from Content-Range, not Content-Length.

Returns:

The file size in bytes, or 0 if the size could not be determined.

Return type:

int

model_post_init(_File__context: Any) → None

Fetch remote metadata if size or modify date is missing.

If both api_size and last_modified are falsy, schedules a background task to fetch metadata from the remote server.

Parameters:

__context (Any) – Pydantic validation context (unused).

property modify: datetime

Return the last modification date.

Returns:

The last modification datetime.

Return type:

datetime

Raises:

ValueError – If the modification date has not been set.

property month: int | None

Return the inferred month from metadata.

Returns:

The month if present in metadata, otherwise None.

Return type:

int or None

record: Recurso

property size: int

Return the file size in bytes.

Returns:

The file size, or 0 if unknown.

Return type:

int

property state: Annotated[str, AfterValidator(func=_validate_state)] | None

Return the inferred state from metadata.

Returns:

The state abbreviation if present in metadata, otherwise None.

Return type:

State or None

type: str

property year: int | None

Return the inferred year from metadata.

Returns:

The year if present in metadata, otherwise None.

Return type:

int or None

class pysus.api.dadosgov.models.Group(record: ConjuntoDados, dataset: BaseRemoteDataset, formatter: Callable | None = None)

Bases: BaseRemoteGroup

A group of files within a dataset.

property description: str

Return an empty description for the group.

Returns:

An empty string.

Return type:

str

extractor_types: ClassVar[list] = [<class 'pysus.api.dadosgov.metadata.DadosGovGroupExtractor'>]

Extractor classes the concrete subclass wants applied. Each client module (ftp/dadosgov/ducklake/saude) overrides this.

property long_name: str

Return the group title.

Returns:

The title of the underlying API record.

Return type:

str

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self – The BaseModel instance.

• context – The context.

property name: str

Return the group name, resolved through dataset aliases.

Returns:

The alias for the group slug if defined, otherwise the raw slug.

Return type:

str

record: ConjuntoDados

Package Root

PySUS — Python interface to Brazilian public health datasets.

PySUS provides seamless access to Brazil’s DATASUS and open-health data repositories, covering disease notifications (SINAN), vital statistics (SINASC, SIM), hospital admissions (SIH), ambulatory care (SIA), immunisations (PNI), health facilities (CNES), and 18 Saude portal themes.

Two styles of public API ─────────────────────────

Flat (original):

import pysus
pysus.sinan(disease="deng", year=2020, as_dataframe=True)

Origin-namespaced (recommended):

pysus.ftp.sinan(disease="deng", year=2020, as_dataframe=True)
pysus.saude.arboviroses(as_dataframe=True)
pysus.dadosgov.sinan(disease="deng", year=2020, as_dataframe=True)

The namespaced style makes the data source impossible to ignore, preventing the class of bug where a call silently returns a different dataset snapshot than the caller expected.

Origins ─────── - pysus.ftp — DATASUS FTP (S3 catalog mirror; 10 databases) - pysus.dadosgov — dados.gov.br (CKAN portal; 6 databases) - pysus.saude — dadosabertos.saude.gov.br (18 theme datasets)

Each namespace exposes the same interface: per-database fetchers, discovery (list_files, info), and metadata (get_origin_meta).

Quick-start ─────────── >>> import pysus >>> pysus.info() # show all available datasets >>> pysus.set_cache(“/my/cache”) # change the download cache path >>> df = pysus.sinan(disease=”deng”, year=2020, as_dataframe=True)

Origin namespace examples ───────────────────────── >>> pysus.ftp.sih(state=”RJ”, year=2020, month=1, as_dataframe=True) >>> pysus.dadosgov.sinasc(state=”SP”, year=2019, as_dataframe=True) >>> pysus.saude.vacinacao(as_dataframe=True)

Getting help ──────────── >>> import pysus >>> pysus.ftp.sinan? # help on a specific function >>> pysus.ftp.info() # datasets for the FTP origin >>> pysus.saude.get_origin_meta() # metadata about the Saude origin

exception pysus.AuthenticationError(message: str = '', *, hint: str = '', docs_url: str = '')

Authentication or authorization failure (missing/wrong credentials).

hint: str = 'Set DADOSGOV_TOKEN env var or pass token= to the client.'

exception pysus.CatalogError(message: str = '', *, hint: str = '', docs_url: str = '')

DuckLake catalog database operation failure (connect, query, schema).

hint: str = 'Ensure the DuckLake catalog is initialized (pysus ducklake init).'

class pysus.ColumnInfo(name: str, description: str = '', description_en: str = '', dtype: str = '', dataset: str = '', endpoint: str = '', categories: str = '', characteristics: str = '', required: bool = False, format: str = '')

Metadata for a single column in a SUS dataset.

name

Column name as it appears in the data file (lowercase).

Type:

str

description

Human-readable description (Portuguese).

Type:

str

description_en

English description, or empty string if unavailable.

Type:

str

dtype

Data type string (e.g. "integer", "string", "VARCHAR(1)").

Type:

str

dataset

Parent dataset name (e.g. "arboviroses", "sinan").

Type:

str

endpoint

Sub-grouping within the dataset (e.g. "dengue", or the SINAN disease form such as "peste").

Type:

str

categories

Value codes of the field (e.g. "1-Sim 2-Não 9-Ignorado"), or empty when unavailable.

Type:

str

characteristics

Field rules and notes from the official dictionary (obligatory, dependencies, validation), or empty when unavailable.

Type:

str

required

Whether the field is mandatory on the notification form.

Type:

bool

format

Expected format hint (e.g. "YYYYMMDD"), if any.

Type:

str

categories: str

characteristics: str

dataset: str

description: str

description_en: str

dtype: str

endpoint: str

format: str

name: str

required: bool

exception pysus.ConnectionError(message: str = '', *, hint: str = '', docs_url: str = '')

Failed to connect to a data source (DuckLake, FTP, DadosGov).

hint: str = 'Check your network connection and verify the data source is available.'

exception pysus.ConversionError(message: str = '', *, hint: str = '', docs_url: str = '')

Error converting between file formats (e.g. DBF -> Parquet).

hint: str = 'The source file may be corrupted or in an unsupported format.'

class pysus.DetectedUnit(column: str, unit: str, confidence: float, detected_from: Literal['column_name', 'value_range', 'metadata'])

Detected unit for a column.

column

Column name.

Type:

str

unit

Detected unit (e.g. "kg", "mg/dL").

Type:

str

confidence

Detection confidence 0.0-1.0.

Type:

float

detected_from

Detection method used.

Type:

str

column: str

confidence: float

detected_from: Literal['column_name', 'value_range', 'metadata']

unit: str

exception pysus.DownloadError(message: str = '', *, hint: str = '', docs_url: str = '')

File download failure (HTTP, S3, or FTP transfer error).

hint: str = 'Check your network connection and verify the file exists.'

exception pysus.FormatError(message: str = '', *, hint: str = '', docs_url: str = '')

File format not recognized or not supported.

hint: str = 'Ensure the file is a supported format (Parquet, DBF, CSV, etc.).'

exception pysus.ParseError(message: str = '', *, hint: str = '', docs_url: str = '')

Error parsing file contents or file metadata.

hint: str = 'The file may be corrupted. Try re-downloading.'

class pysus.PartialDownload(dest_path: Path)

Manages partial file downloads with resume support.

Parameters:

dest_path (Path) – Final destination path for the downloaded file.

exists() → bool

Check if a partial download exists.

async resume(url: str, client: AsyncClient, callback: Callable | None = None) → Path

Resume a partial download using HTTP Range.

Parameters:

• url (str) – URL to download from.

• client (httpx.AsyncClient) – HTTP client to use.

• callback (callable, optional) – Progress callback (downloaded, total) -> None.

Returns:

Path to the completed file.

Return type:

Path

Notes

If the server does not support Range requests, falls back to a fresh download.

size() → int

Return the size of the partial file in bytes.

async start(url: str, client: AsyncClient, callback: Callable | None = None) → Path

Start a fresh download.

Parameters:

• url (str) – URL to download from.

• client (httpx.AsyncClient) – HTTP client to use.

• callback (callable, optional) – Progress callback (downloaded, total) -> None.

Returns:

Path to the completed file.

Return type:

Path

class pysus.PySUS(db_path: Path = PosixPath('/home/docs/pysus/config.db'))

Central orchestrator for downloading and querying PySUS datasets.

download(file: BaseRemoteFile, token: str | None = None, callback: Callable | None = None, timeout: float | None = None) → BaseLocalFile

Download a remote file and return a local file handle.

Skips re-download if a matching local copy already exists. When no callback is supplied and progress bars are enabled, a tqdm bar is shown automatically.

Parameters:

• file (BaseRemoteFile) – The remote file to download.

• token (str, optional) – Access token for authenticated clients (e.g. DadosGov).

• callback (Callable, optional) – Progress callback invoked during the download.

• timeout (float, optional) – Maximum seconds to wait for the download. None (default) means no timeout.

Returns:

The downloaded file wrapped in the appropriate handler.

Return type:

BaseLocalFile

Raises:

• ValueError – If the file’s client is not recognised.

• RuntimeError – If the download fails for any reason.

download_to_parquet(file: BaseRemoteFile, token: str | None = None, callback: Callable[[int, int], None] | None = None, timeout: float | None = None, add_dv: bool = True) → Parquet

Download a file and convert it to Parquet format.

Parameters:

• file (BaseRemoteFile) – The remote file to download and convert.

• token (str, optional) – Access token for authenticated clients.

• callback (Callable[[int, int], None], optional) – Progress callback.

• timeout (float, optional) – Maximum seconds to wait for the download.

• add_dv (bool, optional) – Whether to apply the IBGE verification digit on load (default True).

Returns:

The converted Parquet file handler.

Return type:

Parquet

Raises:

NotImplementedError – If the downloaded file type cannot be converted to Parquet.

get_completed_remote_paths() → set[str]

Return remote paths for all successfully downloaded files.

get_dadosgov(access_token: str | None) → DadosGov

Return the DadosGov client, connecting lazily if needed.

get_ducklake(callback: Callable[[int, int], None] | None = None) → DuckLake

Return the DuckLake client, initializing it lazily if needed.

get_ftp() → FTP

Return the FTP client, connecting lazily if needed.

get_local_file(file: BaseRemoteFile) → BaseLocalFile | None

Look up a previously downloaded file by its remote path.

get_local_hierarchy()

Build a nested dict of cached files grouped by client and dataset.

Returns:

Nested dict keyed by {client: {dataset: {group: [files]}}}.

Return type:

dict

get_saude() → SaudeClient

Return the Saude (dadosabertos.saude.gov.br) client.

No authentication is required; the client is lazily created.

query(client: Annotated[str, AfterValidator(func=_validate_origin)] | None = None, dataset: str | list[str] | None = None, group: str | list[str] | None = None, state: str | list[str] | None = None, year: int | list[int] | None = None, month: int | list[int] | None = None, as_dataframe: bool = False, columns: list[str] | None = None, dtypes: dict[str, str] | None = None) → list[BaseRemoteFile] | DataFrame

Query available datasets through the DuckLake catalog.

Parameters:

• client (Origin, optional) – Source client to filter by.

• dataset (str or list of str, optional) – Dataset name(s) to filter by.

• group (str or list of str, optional) – Group name pattern(s) to filter by (case-insensitive ILIKE).

• state (str or list of str, optional) – Two-letter state code(s) to filter by.

• year (int or list of int, optional) – Year(s) to filter by.

• month (int or list of int, optional) – Month(s) to filter by.

• as_dataframe (bool, optional) – When True, downloads all matching files, converts them to Parquet, and returns a single concatenated DataFrame. Default False.

• columns (list of str, optional) – Subset of column names to keep when as_dataframe=True.

• dtypes (dict, optional) – Override column types when as_dataframe=True. Keys are column names, values are dtype strings.

Returns:

List of matching File objects, or a DataFrame if as_dataframe=True.

Return type:

list or DataFrame

read_parquet(paths: list[Path], sql: str | None = None, mode: Literal['union', 'intersection', 'strict'] = 'union', add_dv: bool = True) → DuckDBPyConnection | pd.DataFrame

Read Parquet files with optional schema handling and SQL filter.

Parameters:

• paths (list of Path) – One or more Parquet file paths to read.

• sql (str, optional) – Optional SQL filter expression applied to the result.

• mode ({"union", "intersection", "strict"}, optional) – Schema resolution mode (default "union").

• add_dv (bool, optional) – When True, automatically applies the IBGE verification digit to municipality code columns. If matching columns are found, a DataFrame is returned instead of a DuckDBPyConnection.

Returns:

The query result.

Return type:

DuckDBPyConnection or pd.DataFrame

Raises:

ValueError – If no paths are provided, or if the schema mode is "strict" and the files have differing schemas.

exception pysus.PySUSError(message: str = '', *, hint: str = '', docs_url: str = '')

Base exception for all PySUS errors.

docs_url: str = 'https://pysus.readthedocs.io/en/latest/errors'

hint: str = ''

exception pysus.PySUSWarning

Non-fatal warning for degraded operation.

exception pysus.ValidationError(message: str = '', *, hint: str = '', docs_url: str = '')

Input validation error (missing fields, invalid values).

hint: str = 'Check the function signature and required parameters.'

pysus.aggregate_by_age_group(df: DataFrame, age_col: str, value_col: str, bins: list[int] | None = None) → DataFrame

Aggregate data by age groups.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• age_col (str) – Age column name.

• value_col (str) – Column to aggregate.

• bins (list, optional) – Age group boundaries. Default: [0, 5, 15, 30, 50, 70, 120].

Returns:

Age group aggregation.

Return type:

pd.DataFrame

pysus.aggregate_by_period(df: DataFrame, date_col: str, value_col: str, freq: Literal['M', 'Q', 'Y'] = 'M') → DataFrame

Aggregate data by time period.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• date_col (str) – Date column (YYYYMMDD format).

• value_col (str) – Column to aggregate.

• freq (str) – Frequency (``”M”``=monthly, ``”Q”``=quarterly, ``”Y”``=yearly).

Returns:

Period aggregation.

Return type:

pd.DataFrame

pysus.aggregate_by_state(df: DataFrame, value_col: str, agg_func: Literal['count', 'sum', 'mean', 'median'] = 'count') → DataFrame

Aggregate data by state (UF).

Parameters:

• df (pd.DataFrame) – Input DataFrame with UF column.

• value_col (str) – Column to aggregate.

• agg_func (str) – Aggregation function.

Returns:

State-level aggregation.

Return type:

pd.DataFrame

pysus.arboviroses(**kwargs) → list[str] | DataFrame

Fetch arboviroses (dengue, chikungunya, zika, yellow fever) from Saude.

Parameters:

**kwargs – Forwarded to _fetch_data().

Examples

>>> pysus.arboviroses(as_dataframe=True)

pysus.assistencia_saude(**kwargs) → list[str] | DataFrame

Fetch hospital and health facility data from Saude.

Examples

>>> pysus.assistencia_saude(as_dataframe=True)

pysus.atencao_primaria(**kwargs) → list[str] | DataFrame

Fetch primary care data (Previne Brasil, SISAB) from Saude.

Examples

>>> pysus.atencao_primaria(as_dataframe=True)

pysus.available_databases() → list[str]

Return list of databases with column metadata.

pysus.bnafar(**kwargs) → list[str] | DataFrame

Fetch pharmaceutical assistance (Hórus stock) from Saude.

Examples

>>> pysus.bnafar(as_dataframe=True)

pysus.cache_status(cache_path: Path | None = None) → CacheStatus

Get status of the cache directory.

Parameters:

cache_path (Path, optional) – Cache directory. If None, uses default.

Returns:

Cache statistics.

Return type:

CacheStatus

pysus.ciencia_tecnologia(**kwargs) → list[str] | DataFrame

Fetch science & technology data (Conitec, RIPSA) from Saude.

Examples

>>> pysus.ciencia_tecnologia(as_dataframe=True)

pysus.ciha(state: Annotated[str, AfterValidator(func=_validate_state)], year: int | list[int], month: int | list[int], group: str | None = 'CIHA', **kwargs) → list[str] | DataFrame

Fetch CIHA hospitalisation records for state, year, month.

CIHA provides hospitalisation records.

Parameters:

• state (str) – Two-letter state abbreviation (e.g. "RJ").

• year (int | list[int]) – Year or list of years to fetch.

• month (int | list[int]) – Month or list of months to fetch.

• group (str, optional) – Grouping code. Default is "CIHA".

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.ciha("RJ", 2020, 1, as_dataframe=True)

pysus.clear_cache(cache_path: Path | None = None) → int

Clear all cached files.

Parameters:

cache_path (Path, optional) – Cache directory. If None, uses default.

Returns:

Number of files removed.

Return type:

int

pysus.cnes(state: Annotated[str, AfterValidator(func=_validate_state)], year: int | list[int], month: int | list[int], group: str | None = None, **kwargs) → list[str] | DataFrame

Fetch CNES health facilities for a state, year, month.

CNES is the Brazilian registry of health-care facilities.

Parameters:

• state (str) – Two-letter state abbreviation (e.g. "RJ").

• year (int | list[int]) – Year or list of years to fetch.

• month (int | list[int]) – Month or list of months to fetch.

• group (str, optional) – Additional grouping code.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.cnes("RJ", 2020, 1, as_dataframe=True)

pysus.column_stats(df: DataFrame) → DataFrame

Get detailed statistics for each column.

Parameters:

df (pd.DataFrame) – Input DataFrame.

Returns:

DataFrame with columns: column, dtype, null_count, null_pct, unique_count, unique_pct, memory_bytes, memory_mb, sample_value. Sorted by memory usage (descending).

Return type:

pd.DataFrame

pysus.covid19(**kwargs) → list[str] | DataFrame

Fetch COVID-19 confirmed cases from dados.gov.br.

Parameters:

**kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.covid19(as_dataframe=True)

pysus.detect_units(df: DataFrame, metadata: dict[str, dict[str, Any]] | None = None) → list[DetectedUnit]

Detect units of measurement for numeric columns.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• metadata (dict, optional) – Column metadata with "unit" keys.

Returns:

Detected units per column.

Return type:

list[DetectedUnit]

pysus.diagnosticos_tratamentos(**kwargs) → list[str] | DataFrame

Fetch diagnostics & treatment protocols (PCDT) from Saude.

Examples

>>> pysus.diagnosticos_tratamentos(as_dataframe=True)

pysus.diff_dfs(df_old: DataFrame, df_new: DataFrame, key_cols: list[str] | None = None) → DiffResult

Compare two DataFrames and return detailed diff.

Parameters:

• df_old (pd.DataFrame) – Older/previous version.

• df_new (pd.DataFrame) – Newer/current version.

• key_cols (list, optional) – Columns identifying rows for matching. If None, compares by position.

Returns:

Detailed diff result.

Return type:

DiffResult

pysus.diff_rows(df_old: DataFrame, df_new: DataFrame, key_cols: list[str] | None = None) → DataFrame

Return rows that differ between two DataFrames.

Parameters:

• df_old (pd.DataFrame) – Older version.

• df_new (pd.DataFrame) – Newer version.

• key_cols (list, optional) – Columns for matching rows.

Returns:

DataFrame with diff info columns.

Return type:

pd.DataFrame

pysus.diff_summary(df_old: DataFrame, df_new: DataFrame) → dict

Quick summary diff without detailed cell-level comparison.

Parameters:

• df_old (pd.DataFrame) – Older version.

• df_new (pd.DataFrame) – Newer version.

Returns:

Summary of changes.

Return type:

dict

pysus.disable_progress_bars() → None

Disable tqdm progress bars for downloads globally.

async pysus.download_many(files: list[Any], download_fn: Callable, max_workers: int = 4, callback: Callable | None = None) → list[Any]

Download multiple files concurrently.

Parameters:

• files (list) – List of file objects to download.

• download_fn (Callable) – Async function that downloads a single file. Signature: async def download_fn(file, callback) -> result.

• max_workers (int) – Maximum concurrent downloads.

• callback (Callable, optional) – Progress callback (downloaded, total) -> None. Called with aggregate progress across all files.

Returns:

Results from each download (same order as input files).

Return type:

list

pysus.economia_saude(**kwargs) → list[str] | DataFrame

Fetch health economics data (BPS, ApuraSUS, SIOPS) from Saude.

Examples

>>> pysus.economia_saude(as_dataframe=True)

pysus.educacao_saude(**kwargs) → list[str] | DataFrame

Fetch health education data (PVC) from Saude.

Examples

>>> pysus.educacao_saude(as_dataframe=True)

pysus.enable_progress_bars() → None

Enable tqdm progress bars for downloads globally.

pysus.export(df: DataFrame, path: str | Path, **kwargs: object) → Path

Export DataFrame to a file, auto-detecting format from extension.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• path (str or Path) – Output file path.

• **kwargs – Additional arguments passed to the format-specific writer.

Returns:

Path to created file.

Return type:

Path

Raises:

• ValueError – If the file extension is not supported.

• Supported Formats –

• ----------------- –

• - .csv — CSV with metadata sidecar –

• - .xlsx — Excel with metadata sheet –

• - .parquet — Parquet (native pandas) –

• - .json — JSON lines –

• - .sqlite — SQLite table (requires table_name kwarg) –

pysus.flatten_json_columns(df: pd.DataFrame, columns: list[str] | None = None) → pd.DataFrame

Flatten columns containing JSON strings into flat columns.

For each column specified (or auto-detected), parses JSON strings and expands them into new columns with a <original>.<key> naming pattern.

Parameters:

• df (DataFrame) – Input DataFrame.

• columns (list of str, optional) – Specific columns to flatten. When None (default), auto-detects columns where a sample of non-null values looks like JSON objects.

Returns:

A copy with JSON columns expanded into flat columns and the original JSON columns dropped.

Return type:

DataFrame

Notes

• Nested lists are kept as-is (not recursively flattened).

• JSON objects with differing keys across rows are unioned; missing keys become NaN.

• Non-JSON values in a column are left as-is (column is not dropped).

pysus.get_aliases(database: str, column: str) → list[str]

Get historical aliases for a column name.

Parameters:

• database (str) – Database name ("sinan", "sih", etc.).

• column (str) – Current column name.

Returns:

Previous names for this column.

Return type:

list[str]

pysus.get_linking_keys(database1: str, database2: str) → list[str]

Get common linking keys between two databases.

Parameters:

• database1 (str) – First database name.

• database2 (str) – Second database name.

Returns:

Column names that can link the two datasets.

Return type:

list[str]

pysus.ibge(year: int | list[int], group: str | None = None, **kwargs) → list[str] | DataFrame

Fetch IBGE census data for given year(s).

IBGE provides census and demographic data.

Parameters:

• year (int | list[int]) – Year or list of years to fetch.

• group (str, optional) – Additional grouping code.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.ibge(2020, as_dataframe=True)

pysus.info_table() → None

Print a table of all available datasets across every origin.

This is the primary entry-point for a new user:

import pysus
pysus.info()

Each row suggests the namespaced call for that dataset. Those fetchers return a FileBag by default; pass as_dataframe=True for a pandas.DataFrame, or download=False to list the files without fetching anything.

pysus.link_datasets(df1: DataFrame, df2: DataFrame, on: str | list[str], how: Literal['inner', 'left', 'right', 'outer'] = 'inner') → DataFrame

Link two DATASUS datasets on common keys.

Handles column name conflicts by adding suffixes.

Parameters:

• df1 (pd.DataFrame) – First DataFrame.

• df2 (pd.DataFrame) – Second DataFrame.

• on (str or list) – Column name(s) to join on.

• how (str) – Join type.

Returns:

Merged DataFrame.

Return type:

pd.DataFrame

pysus.list_files(dataset: Annotated[str, AfterValidator(func=_validate_dataset_name)], client: Annotated[str, AfterValidator(func=_validate_origin)] | None = None, group: str | None = None, state: str | None = None, year: int | list[int] | None = None, month: int | list[int] | None = None, **kwargs) → DataFrame

List catalog files filtered by client, group, state, year, and month.

Queries the PySUS API metadata and returns a DataFrame with file data without downloading the actual files.

Parameters:

• dataset (str) – Dataset name (e.g. "SINAN").

• client (str, optional) – Data source client to query.

• group (str, optional) – Group or disease code to filter by.

• state (str, optional) – Two-letter state abbreviation.

• year (int | list[int], optional) – Year or list of years to filter by.

• month (int | list[int], optional) – Month or list of months to filter by.

• **kwargs – Forwarded to PySUS.query().

Returns:

Columns: name, path, dataset, group, year, month, state, modify.

Return type:

pd.DataFrame

Examples

>>> pysus.list_files("SINAN", year=2020, state="RJ")

pysus.load_column_metadata(database: str, group: str | None = None) → dict[str, dict[str, Any]]

Load column metadata for a DATASUS database.

Parameters:

• database (str) – Database name: "sinan", "sih", "sia", "sim", "sinasc".

• group (str, optional) – Sub-group within the database (e.g. "arboviroses" for SINAN). When None, returns metadata for all groups.

Returns:

Mapping of column names (uppercase) to their definitions:

{
    'DT_NOTIFIC': {
        'type': 'string',
        'description_pt': 'Data da notificação',
        'description_en': 'Notification date',
        'format': 'YYYYMMDD',
        'required': True
    }
}

Return type:

dict[str, dict]

pysus.macro_saude(**kwargs) → list[str] | DataFrame

Fetch macro-region and health region data from Saude.

Examples

>>> pysus.macro_saude(as_dataframe=True)

pysus.mask_data(df: DataFrame, columns: list[str] | None = None, key: bytes | None = None, method: Literal['encrypt', 'hash', 'redact'] = 'encrypt') → tuple[DataFrame, bytes]

Mask sensitive data in DataFrame.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• columns (list, optional) – Columns to mask (auto-detect if None).

• key (bytes, optional) – Encryption key (generated if None).

• method (str) – Masking method.

Returns:

Masked DataFrame and key for reversal.

Return type:

tuple[pd.DataFrame, bytes]

pysus.missing_values(df: DataFrame, group_by: str | None = None, threshold: float = 0.0) → DataFrame | tuple[DataFrame, DataFrame]

Analyze missing values in a DataFrame.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• group_by (str, optional) – Column to group by (e.g. "UF" for state-level analysis).

• threshold (float) – Minimum missing percentage to include (0.0-1.0).

Returns:

DataFrame with columns: column, missing_count, missing_pct, complete_count, complete_pct. If group_by is provided, returns a tuple of (summary_df, grouped_df).

Return type:

pd.DataFrame or tuple

pysus.optimize_memory(df: DataFrame, downcast_int: bool = True, downcast_float: bool = True) → DataFrame

Optimize memory usage by downcasting numeric types.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• downcast_int (bool) – Downcast integer columns.

• downcast_float (bool) – Downcast float columns.

Returns:

Memory-optimized DataFrame.

Return type:

pd.DataFrame

pysus.outros_temas(**kwargs) → list[str] | DataFrame

Fetch miscellaneous CED coordination data from Saude.

Examples

>>> pysus.outros_temas(as_dataframe=True)

pysus.ouvidoria(**kwargs) → list[str] | DataFrame

Fetch SUS ombudsman complaints from Saude.

Examples

>>> pysus.ouvidoria(as_dataframe=True)

pysus.pda(**kwargs) → list[str] | DataFrame

Fetch digital health and open data plan data from Saude.

Examples

>>> pysus.pda(as_dataframe=True)

pysus.pni(state: Annotated[str, AfterValidator(func=_validate_state)], year: int | list[int], group: str | None = None, **kwargs) → list[str] | DataFrame

Fetch PNI immunisation records for a given state, year(s).

PNI is the Brazilian national immunisation programme.

Parameters:

• state (str) – Two-letter state abbreviation (e.g. "RJ").

• year (int | list[int]) – Year or list of years to fetch.

• group (str, optional) – Additional grouping code.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.pni("RJ", 2020, as_dataframe=True)

pysus.prevencao_promocao(**kwargs) → list[str] | DataFrame

Fetch prevention & promotion (EPI distribution) from Saude.

Examples

>>> pysus.prevencao_promocao(as_dataframe=True)

pysus.profile_report(df: DataFrame, output: str | Path | None = None, format: Literal['html', 'json', 'text'] = 'text') → str | dict[str, Any]

Generate comprehensive profiling report.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• output (str or Path, optional) – File path to save report.

• format (str) – Output format: "html", "json", or "text".

Returns:

Report as string (text/html) or dict (json).

Return type:

str or dict

pysus.quality_score(df: DataFrame, schema: dict[str, dict[str, Any]] | None = None) → QualityScore

Calculate overall quality score for a DataFrame.

Scoring weights: - Completeness: 40% (average non-null percentage) - Validity: 40% (values passing basic validation) - Consistency: 20% (values matching expected patterns)

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• schema (dict, optional) – Column metadata for validation (from load_column_metadata).

Returns:

Quality score with overall and component scores.

Return type:

QualityScore

pysus.query_parquet(path: str | Path, query: str = 'SELECT * FROM data') → DuckDBPyRelation

Query Parquet files as a DuckDB relation.

Parameters:

• path (str or Path) – Path to a single Parquet file or glob pattern.

• query (str) – SQL query. The table name data refers to the Parquet file(s).

Returns:

Lazy relation for further processing.

Return type:

duckdb.DuckDBPyRelation

Examples

>>> rel = query_parquet("data.parquet")
>>> rel = query_parquet("data.parquet", "SELECT COUNT(*) FROM data")

pysus.rename_columns(df: DataFrame, mapping: dict[str, str] | None = None, database: str | None = None) → DataFrame

Rename columns using mapping or historical aliases.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• mapping (dict, optional) – Custom rename mapping {old_name: new_name}.

• database (str, optional) – Database name to use historical aliases.

Returns:

DataFrame with renamed columns.

Return type:

pd.DataFrame

pysus.retry(max_attempts: int = 3, backoff_base: float = 1.0, backoff_max: float = 30.0, exceptions: tuple[type[BaseException], ...] = (<class 'Exception'>,)) → Callable

Decorator that retries a function on transient failures.

Parameters:

• max_attempts (int) – Maximum number of attempts (including the first).

• backoff_base (float) – Base delay in seconds (doubled each retry).

• backoff_max (float) – Maximum delay in seconds.

• exceptions (tuple) – Exception types to catch and retry on.

Returns:

Decorated function.

Return type:

Callable

Examples

>>> @retry(max_attempts=3, exceptions=(ConnectionError,))
... async def fetch(url):
...     ...

pysus.saude_indigena(**kwargs) → list[str] | DataFrame

Fetch indigenous health data (Siasi/SasiSUS/Sesai) from Saude.

Examples

>>> pysus.saude_indigena(as_dataframe=True)

pysus.search(query: str, *, origin: str | None = None) → None

Search datasets and column metadata by keyword.

Parameters:

• query (str) – Search term (e.g. "dengue", "SIH", "peso").

• origin (str, optional) – Restrict to a specific origin ("FTP", "Saude", "DadosGov"). None searches all.

Examples

>>> pysus.search("dengue")
>>> pysus.search("SIH", origin="FTP")
>>> pysus.search("peso")

pysus.search_columns(dataset: str | None = None, query: str | None = None, *, endpoint: str | None = None) → list[ColumnInfo]

Search column metadata across all available schema sources.

Parameters:

• dataset (str, optional) – Restrict to a specific dataset name (e.g. "arboviroses" or "sinan"). When None, searches all datasets.

• query (str, optional) – Case-insensitive substring match against column name and descriptions (both PT and EN). When None, returns all columns for the given dataset.

• endpoint (str, optional) – Restrict to a specific endpoint within the dataset (e.g. "dengue", or a SINAN disease form such as "peste").

Returns:

Matching column metadata, sorted by name.

Return type:

list[ColumnInfo]

Examples

>>> search_columns("arboviroses", "date")
[ColumnInfo(name='dt_notific', description='Data da notificação', ...)]
>>> search_columns("sinan", endpoint="peste")
[ColumnInfo(name='con_classi', description='', ...)]

pysus.set_cache(path: str | Path) → Path

Set the global cache directory for PySUS.

All downloaded files are stored under this path. The default is ~/pysus (or the PYSUS_CACHEPATH environment variable).

Parameters:

path (str or Path) – New cache directory. Parent directories are created automatically.

Returns:

The resolved cache path.

Return type:

Path

Examples

>>> from pysus import set_cache
>>> set_cache("/data/pysus")
PosixPath('/data/pysus')

pysus.set_precision(df: DataFrame, precision: Literal['float16', 'float32', 'float64'] = 'float32', columns: list[str] | None = None) → DataFrame

Set numeric precision for DataFrame columns.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• precision (str) – Target precision.

• columns (list, optional) – Specific columns (all numeric if None).

Returns:

DataFrame with adjusted precision.

Return type:

pd.DataFrame

pysus.sia(state: Annotated[str, AfterValidator(func=_validate_state)], year: int | list[int], month: int | list[int], group: str | None = None, **kwargs) → list[str] | DataFrame

Fetch SIA ambulatory care for a state, year, month.

SIA is the Brazilian ambulatory care information system.

Parameters:

• state (str) – Two-letter state abbreviation (e.g. "RJ").

• year (int | list[int]) – Year or list of years to fetch.

• month (int | list[int]) – Month or list of months to fetch.

• group (str, optional) – Additional grouping code.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.sia("RJ", 2020, 1, as_dataframe=True)

pysus.sih(state: Annotated[str, AfterValidator(func=_validate_state)], year: int | list[int], month: int | list[int], group: str | None = None, **kwargs) → list[str] | DataFrame

Fetch SIH hospital admissions for a state, year, month.

SIH is the Brazilian hospital admission information system.

Parameters:

• state (str) – Two-letter state abbreviation (e.g. "RJ").

• year (int | list[int]) – Year or list of years to fetch.

• month (int | list[int]) – Month or list of months to fetch.

• group (str, optional) – Additional grouping code.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.sih("RJ", 2020, 1, as_dataframe=True)

pysus.sim(state: Annotated[str, AfterValidator(func=_validate_state)], year: int | list[int], group: str | None = None, **kwargs) → list[str] | DataFrame

Fetch SIM mortality records for a given state, year(s).

SIM is the Brazilian mortality information system.

Parameters:

• state (str) – Two-letter state abbreviation (e.g. "RJ").

• year (int | list[int]) – Year or list of years to fetch.

• group (str, optional) – Additional grouping code.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.sim("RJ", 2020, as_dataframe=True)

pysus.sinan(disease: Annotated[str, AfterValidator(func=_validate_dataset_name)], year: int | list[int], **kwargs) → list[str] | DataFrame

Fetch SINAN records for a given disease and year(s).

SINAN is the Brazilian notifiable-disease information system.

Parameters:

• disease (str) – Disease code (e.g. "DENG" for dengue, "ZIKA").

• year (int | list[int]) – Year or list of years to fetch.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.sinan("DENG", 2020, as_dataframe=True)

pysus.sinasc(state: Annotated[str, AfterValidator(func=_validate_state)], year: int | list[int], group: str | None = None, **kwargs) → list[str] | DataFrame

Fetch SINASC birth certificates for a given state, year(s).

SINASC is the Brazilian live birth information system.

Parameters:

• state (str) – Two-letter state abbreviation (e.g. "RJ").

• year (int | list[int]) – Year or list of years to fetch.

• group (str, optional) – Additional grouping code.

• **kwargs – Forwarded to _fetch_data().

Returns:

Paths or DataFrame.

Return type:

list[str] | pd.DataFrame

Examples

>>> pysus.sinasc("RJ", 2020, as_dataframe=True)

pysus.sisagua(**kwargs) → list[str] | DataFrame

Fetch water quality surveillance data from Saude.

Examples

>>> pysus.sisagua(as_dataframe=True)

pysus.sisvan(**kwargs) → list[str] | DataFrame

Fetch food & nutrition surveillance data from Saude.

Examples

>>> pysus.sisvan(as_dataframe=True)

pysus.stream_parquet(path: str | Path, chunk_size: int = 10000, columns: list[str] | None = None) → Generator[DataFrame, None, None]

Stream large Parquet files in chunks.

Parameters:

• path (str or Path) – Path to Parquet file.

• chunk_size (int) – Number of rows per chunk.

• columns (list, optional) – Columns to read.

Yields:

pd.DataFrame – DataFrame chunks.

pysus.to_arrow(rel: DuckDBPyRelation)

Export a DuckDB relation to an Arrow table.

Parameters:

rel (duckdb.DuckDBPyRelation) – The relation to export.

Returns:

The result as an Arrow table.

Return type:

pyarrow.Table

pysus.to_csv(df: DataFrame, path: str | Path, metadata: dict | None = None, encoding: str = 'utf-8') → Path

Export DataFrame to CSV with optional metadata sidecar.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• path (str or Path) – Output file path.

• metadata (dict, optional) – Metadata to save in .metadata.json sidecar file.

• encoding (str) – File encoding.

Returns:

Path to created file.

Return type:

Path

pysus.to_df(rel: DuckDBPyRelation)

Materialize a DuckDB relation to a pandas DataFrame.

Parameters:

rel (duckdb.DuckDBPyRelation) – The relation to materialize.

Returns:

The result as a DataFrame.

Return type:

pd.DataFrame

pysus.to_english(df: DataFrame, store_mapping: bool = True) → DataFrame

Rename Portuguese column names to English equivalents.

Parameters:

• df (DataFrame) – Input DataFrame with Portuguese column names.

• store_mapping (bool, optional) – When True (default), stores the original-to-English mapping in df.attrs["aliases"] so it can be reversed later.

Returns:

A copy with English column names.

Return type:

DataFrame

Examples

>>> import pysus
>>> df = pysus.to_english(df)
>>> df.attrs["aliases"]  # original → English mapping
{'DT_NOTIFIC': 'notification_date', ...}

pysus.to_excel(df: DataFrame, path: str | Path, metadata: dict | None = None, sheet_name: str = 'Data') → Path

Export DataFrame to Excel with metadata sheet.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• path (str or Path) – Output file path.

• metadata (dict, optional) – Metadata to include as a separate sheet.

• sheet_name (str) – Name of the data sheet.

Returns:

Path to created file.

Return type:

Path

pysus.to_geojson(df: DataFrame, path: str | Path, lat_col: str = 'LATITUDE', lon_col: str = 'LONGITUDE', geocode_col: str | None = None, properties: list[str] | None = None) → Path

Export DataFrame to GeoJSON with Point geometries.

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• path (str or Path) – Output file path.

• lat_col (str) – Latitude column name.

• lon_col (str) – Longitude column name.

• geocode_col (str, optional) – Geocode column for properties.

• properties (list, optional) – Additional columns to include as properties.

Returns:

Path to created file.

Return type:

Path

pysus.to_sql(df: DataFrame, table_name: str, dialect: Literal['duckdb', 'mysql', 'postgresql', 'sqlite'] = 'duckdb', include_data: bool = False) → str

Generate SQL CREATE TABLE statement from DataFrame.

Parameters:

• df (pd.DataFrame) – Input DataFrame (used for schema inference).

• table_name (str) – Name for the SQL table.

• dialect (str) – SQL dialect.

• include_data (bool) – Include INSERT statements.

Returns:

SQL DDL string.

Return type:

str

pysus.unmask_data(df: DataFrame, columns: list[str], key: bytes) → DataFrame

Reverse encryption masking.

Parameters:

• df (pd.DataFrame) – DataFrame with masked columns.

• columns (list) – Columns to unmask.

• key (bytes) – Encryption key used for masking.

Returns:

Unmasked DataFrame.

Return type:

pd.DataFrame

pysus.vacinacao(**kwargs) → list[str] | DataFrame

Fetch vaccination data (PNI doses, ESAVI) from Saude.

Examples

>>> pysus.vacinacao(as_dataframe=True)

pysus.validate_choice(value: str, choices: list[str], label: str = 'value', cutoff: float = 0.6) → str

Validate that value is in choices, suggesting corrections.

Parameters:

• value (str) – User-provided value.

• choices (list[str]) – Allowed values.

• label (str) – Human-readable name for the parameter (for the error message).

• cutoff (float) – Similarity threshold for suggestions (0-1).

Returns:

The matched choice (case-insensitive match returned as canonical).

Return type:

str

Raises:

ValidationError – If value is not in choices and no close match is found.

pysus.validate_data(df: DataFrame, rules: dict[str, dict[str, Any]] | None = None) → list[ValidationResult]

Validate DataFrame against built-in and custom rules.

Built-in rules: - Age columns (IDADE, NU_IDADE_N): 0-120 - Date columns (DT_*): YYYYMMDD format, reasonable range - Categorical columns (CS_*): values in expected set

Parameters:

• df (pd.DataFrame) – Input DataFrame.

• rules (dict, optional) – Custom validation rules per column. Keys are column names, values are dicts with "type" and parameters.

Returns:

Validation results (includes passing rules).

Return type:

list[ValidationResult]

pysus.validate_dataset(name: str) → str

Validate a PySUS dataset name.

Parameters:

name (str) – Dataset name (e.g. ‘sinan’, ‘SIH’).

Returns:

Canonical dataset name.

Return type:

str

Raises:

ValidationError – If dataset name is not recognised.

pysus.validate_origin(origin: str) → str

Validate a data origin.

Parameters:

origin (str) – Origin name (e.g. ‘ftp’, ‘dadosgov’).

Returns:

Canonical origin name.

Return type:

str

Raises:

ValidationError – If origin is not recognised.

pysus.validate_source(source: str) → str

Validate a fetch source.

A source is where data is read from. "catalog" (default) serves the DuckLake catalog/S3 mirror; "origin" hits the origin server directly.

Parameters:

source (str) – Source name ("catalog" or "origin").

Returns:

Canonical source name.

Return type:

str

Raises:

ValidationError – If source is not recognised.

pysus.vigilancia_meio_ambiente(**kwargs) → list[str] | DataFrame

Fetch environmental surveillance (SRAG, SIM, mpox) from Saude.

Examples

>>> pysus.vigilancia_meio_ambiente(as_dataframe=True)

pysus.warn(message: str, *, stacklevel: int = 2) → None

Emit a PySUSWarning.

DuckLake Internals

pysus.api.ducklake.functional.alias_marker(target_key: str) → str

Return the alias marker content pointing to target_key.

async pysus.api.ducklake.functional.download_http(remote_path: str, local_path: Path, callback: Callable[[int, int], None] | None = None) → None

async pysus.api.ducklake.functional.upload_s3(local_path: Path, remote_path: str, access_key: str, secret_key: str, callback: Callable[[int, int], None] | None = None) → None

class pysus.api.ducklake.catalog.adapters.BaseAdapter(engine=None, credentials: DuckLakeCredentials | None = None, update_on_close: bool = False, **data)

Bases: ABC

cache_dir: Path = PosixPath('/home/docs/pysus/ducklake')

checkpoint() → None

Force a WAL checkpoint, persisting writes to db_local.

async close(update: bool = False) → None

async connect(force: bool = False, callback: Callable[[int, int], None] | None = None) → None

property connected: bool

True when the database engine is initialized.

db_local: Path

db_remote: Path

async ensure_connected(callback: Callable[[int, int], None] | None = None) → None

Connect the engine if it is not already connected.

The shared engine doubles as the process-lifetime anchor, so the DuckDB instance can never be torn down by other lifecycle paths.

get_session() → Session

property local_dirty: bool

True when the local database has unsaved changes.

mark_dirty() → None

Flag the local database as modified (upload on close).

raw_connection()

Return a raw DuckDB connection to the catalog database.

Raises:

CatalogError – If the engine is not initialized (call ensure_connected() first).

async reconnect() → None

Dispose the shared engine and reinitialize from scratch.

Used after the database file is found broken or right before it is replaced by a fresh download.

property remote_url: str

setup_engine(access_key: str | None = None, secret_key: str | None = None) → Engine

Return the shared engine for this adapter’s database file.

DuckDB keeps one database instance per file per process; opening the file again merely attaches to it, and disposing one engine’s connection tears the shared instance down for everyone else. Adapters therefore share one engine per file (process-wide) that is only disposed via _dispose_shared() — right before the file itself is replaced by a re-download.

sql(query: str, params: dict | None = None) → Result

async sync(update: bool = False) → None

Upload the dirty catalog file, keeping the adapter attached.

Unlike close(), the engine and session factory are left in place, so the local database can keep serving writes right after the upload. This is the checkpoint path: the local file we just wrote is authoritative, so re-downloading it (as close + connect does) would be a needless — and hang-prone — network round-trip. The engine is not torn down, exactly as in close: the shared registry releases it only via _dispose_shared() right before a file is replaced.

transaction()

Yield (connection, cursor) for a committed transaction.

Uses the shared engine’s pooled connection — the only way to guarantee every connection to a DuckDB file carries identical configuration (DuckDB rejects mixed-configuration opens) and to keep the process-lifetime instance anchored. The connection is not closed (it is the shared anchor); the transaction is committed on success and rolled back on error.

class pysus.api.ducklake.catalog.adapters.CatalogAdapter(engine=None, **data)

Bases: BaseAdapter

class pysus.api.ducklake.catalog.adapters.ColumnsAdapter(engine=None, **data)

Bases: BaseAdapter

class pysus.api.ducklake.catalog.adapters.DatasetAdapter(name: str, dataset_id: int, engine=None, **data)

Bases: BaseAdapter

class pysus.api.ducklake.catalog.adapters.DuckLakeCredentials(*, access_key: SecretStr, secret_key: SecretStr)

Bases: BaseModel

access_key: SecretStr

secret_key: SecretStr

Central discovery catalog ORM models — stored in catalog.db.

Tracks only available datasets. File-level metadata lives in per-dataset catalog_<name>.db files defined in .dataset.

class pysus.api.ducklake.catalog.orm.default.Base(**kwargs: Any)

Bases: DeclarativeBase

Base class for central catalog ORM models.

metadata: ClassVar[MetaData] = MetaData()

Refers to the _schema.MetaData collection that will be used for new _schema.Table objects.

See also

Accessing Table and Metadata

registry: ClassVar[_RegistryType] = <sqlalchemy.orm.decl_api.registry object>

Refers to the _orm.registry in use where new _orm.Mapper objects will be associated.

class pysus.api.ducklake.catalog.orm.default.Dataset(**kwargs)

Bases: Base

ORM model for the datasets table — central registry of available datasets.

Parameters:

• id (int, optional) – Primary key (auto-generated by sequence).

• name (str) – Unique short name for the dataset.

• long_name (str) – Human-readable full name.

• description (str, optional) – Optional description of the dataset contents.

description: Mapped[str | None]

id: Mapped[int]

long_name: Mapped[str]

name: Mapped[str]

class pysus.api.ducklake.catalog.orm.dataset.DatasetBase(**kwargs: Any)

Bases: DeclarativeBase

metadata: ClassVar[MetaData] = MetaData()

Refers to the _schema.MetaData collection that will be used for new _schema.Table objects.

See also

Accessing Table and Metadata

registry: ClassVar[_RegistryType] = <sqlalchemy.orm.decl_api.registry object>

Refers to the _orm.registry in use where new _orm.Mapper objects will be associated.

class pysus.api.ducklake.catalog.orm.dataset.File(**kwargs)

Bases: DatasetBase

dataset_id: Mapped[int]

group: Mapped[Group | None]

group_id: Mapped[int | None]

id: Mapped[int]

modified: Mapped[datetime]

month: Mapped[int | None]

origin_modified: Mapped[datetime | None]

origin_path: Mapped[str]

origin_size: Mapped[int]

path: Mapped[str]

rows: Mapped[int]

sha256: Mapped[str | None]

size: Mapped[int]

state: Mapped[str | None]

type: Mapped[str]

year: Mapped[int | None]

class pysus.api.ducklake.catalog.orm.dataset.Group(**kwargs)

Bases: DatasetBase

dataset_id: Mapped[int]

description: Mapped[str | None]

files: Mapped[list[File]]

id: Mapped[int]

long_name: Mapped[str]

name: Mapped[str]

class pysus.api.ducklake.catalog.orm.columns.ColumnDefinition(**kwargs)

Bases: ColumnsBase

dataset_id

description

id

name

nullable

type

class pysus.api.ducklake.catalog.orm.columns.ColumnsBase(**kwargs: Any)

Bases: DeclarativeBase

metadata: ClassVar[MetaData] = MetaData()

Refers to the _schema.MetaData collection that will be used for new _schema.Table objects.

See also

Accessing Table and Metadata

registry: ClassVar[_RegistryType] = <sqlalchemy.orm.decl_api.registry object>

Refers to the _orm.registry in use where new _orm.Mapper objects will be associated.

Legacy Data Readers

Byte-level DBF reader for DATASUS files.

Provides a fast path that reads DBF records at the byte level using numpy’s frombuffer with structured dtypes, avoiding the per-row Python-object overhead of dbfread.

Suitable for large DATASUS files (SIA-PA, PNI, etc.) where dbfread’s row-by-row materialisation is the bottleneck.

class pysus.data.dbf_reader.DBFField(name: str, type_: str, length: int, offset: int)

length: int

name: str

offset: int

type_: str

class pysus.data.dbf_reader.DBFSchema(num_records: int, header_len: int, record_len: int, fields: list[pysus.data.dbf_reader.DBFField] = <factory>)

build_dtype() → dtype

property field_names: list[str]

fields: list[DBFField]

header_len: int

num_records: int

record_len: int

pysus.data.dbf_reader.read_dbf_fast(path: str | Path, columns: list[str] | None = None, encoding: str = 'latin-1') → DataFrame

Read an entire DBF file into a DataFrame using vectorised byte access.

Parameters:

• path (str or Path) – Path to the DBF file.

• columns (list[str], optional) – Subset of columns to read. If None all columns are returned.

• encoding (str) – Text encoding for character fields (default latin-1, the encoding DATASUS uses).

Return type:

pd.DataFrame

pysus.data.dbf_reader.read_dbf_filtered(path: str | Path, column: str, values: list[str], columns: list[str] | None = None, encoding: str = 'latin-1', prefix_match: bool = True) → DataFrame

Read only DBF records where column matches one of values.

The scan is performed at the byte level – unmatched rows are never decoded or materialised.

Parameters:

• path (str or Path)

• column (str) – Column name to filter on.

• values (list[str]) – Target values. For prefix_match=True a 3-char value will match any longer value that starts with it.

• columns (list[str], optional) – Columns to return. If None, all columns are returned.

• encoding (str)

• prefix_match (bool) – If True, a value of length < field length acts as a prefix.

Return type:

pd.DataFrame

pysus.data.dbf_reader.read_dbf_schema(path: str | Path) → DBFSchema

Return the schema of a DBF file without reading records.

pysus.data.dbf_reader.stream_dbf_fast(path: str | Path, chunk_size: int = 100000, encoding: str = 'latin-1') → Iterator[DataFrame]

Stream records from a DBF file in chunks using vectorised byte access.

Parameters:

• path (str or Path)

• chunk_size (int) – Number of rows per chunk.

• encoding (str)

Yields:

pd.DataFrame

CLI

pysus.cli.info()

Show a table of all available datasets.

pysus.cli.open(port: int = <typer.models.OptionInfo object>)

Open PySUS in your web browser (alias for ‘web’).

pysus.cli.search(query: str = <typer.models.ArgumentInfo object>)

Search datasets across all origins (FTP, Saude, DadosGov).

pysus.cli.version()

Print the installed PySUS version.

pysus.cli.web(port: int = <typer.models.OptionInfo object>, share: bool = <typer.models.OptionInfo object>)

Launch the Streamlit web interface.