Build a Multi-Source Research Data API

Build a FastAPI service that extracts data from PACS and SMB shares with ATNA audit logging.

Introduction

30 Second Summary

Hospitals store research data across multiple disconnected systems. Imaging lives in one place, waveform recordings sit on a file share somewhere else, and pulling copies for a research study means navigating each system individually.

In this project, you will build a unified API that extracts data from multiple hospital sources behind a single secure endpoint. Every extraction event gets logged in a healthcare-compliant audit format so you always have a record of what was accessed and by whom.

What You'll Build

A Python FastAPI service that accepts extraction requests over HTTPS and dispatches them to the correct data source adapter, saving retrieved files to a local staging directory.

By the end of this project, you'll have:

  • A running FastAPI service with API key authentication that routes extraction requests to the correct source adapter.
  • A DICOM adapter that retrieves imaging studies from a local test PACS via C-GET and an SMB adapter that copies files from a network share.
  • A full IHE ATNA audit trail that logs every extraction event in DICOM PS3.15 compliant XML format.
  • Secret Mission: Add LDAP/Active Directory authentication for production deployment.

Are there any prerequisites?

You should have DCMTK already installed on your system. DCMTK provides the test PACS and DICOM networking tools you will use throughout this project. No prior experience with DICOM or SMB protocols is required.

Before We Start

This project uses several specialized Python libraries to talk to hospital systems. You need Python 3.13, a virtual environment, and a handful of packages including FastAPI, pynetdicom, and smbprotocol.

In this step, you will install Python, create your project folder, set up a virtual environment, and install all the dependencies your extraction API needs.

In this step, get ready to:
  • Install Python 3.13 and verify it works.
  • Create your project directory with a virtual environment and install dependencies.
  • Open the project in VS Code and confirm all packages are importable.
Install Python 3.13

Python 3.13 is the latest stable release with binary installers for both macOS and Windows. Python 3.12 is now in security-fixes-only mode, so 3.13 gives you full ecosystem compatibility and active support.

  • Open your terminal.
  • Check if Python is already installed by running:
python3 --version

✔️ I see Python 3.13.x

You are all set. Skip ahead to the next substep.

ⓧ I see a different version or command not found

  • Download the Python 3.13 installer from python.org/downloads.
  • Run the installer. On Windows, check the Add Python to PATH box before clicking Install Now.
  • Once the installer finishes, close and reopen your terminal.
  • Run python3 --version again to confirm you see Python 3.13.x.

Still seeing an older version?

On macOS, python3 may point to an older system Python. Try python3.13 --version to use the version you just installed.

On Windows, if python opens the Microsoft Store, you missed the PATH checkbox during install. Rerun the installer and select Modify to add it.

Help me figure out why Python 3.13 isn't recognized in my terminal.

Create the project directory and install dependencies

You will create a dedicated folder for this project, set up a virtual environment to isolate your packages, and install all the libraries the extraction API needs.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder, move into it, and set up a virtual environment by running these commands:
mkdir research-extraction-api
cd research-extraction-api
python3 -m venv venv
  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt.

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. Packages you install here stay separate from your system Python, so different projects can use different versions of the same library without conflicts.

  • Create a requirements.txt file that lists all the dependencies your project needs. Create the file by running:
cat > requirements.txt << 'EOF'
fastapi==0.136.1
uvicorn[standard]==0.34.0
pynetdicom==3.0.4
pydicom==3.0.1
smbprotocol==1.16.1
rfc5424-logging-handler==1.4.3
EOF

What are these packages?

  • fastapi is the web framework for your API endpoints.
  • uvicorn is the ASGI server that runs your FastAPI app.
  • pynetdicom and pydicom handle DICOM networking and file parsing for communicating with the PACS.
  • smbprotocol provides high-level file access to Windows SMB network shares.
  • rfc5424-logging-handler enables RFC 5424 syslog transport for ATNA audit messages.
  • Install all the dependencies by running:
pip install -r requirements.txt

You should see pip download and install all six packages plus their dependencies. The final lines should confirm each package was successfully installed.

Pip install failed?

Make sure your virtual environment is activated. You should see (venv) in your terminal prompt. If not, run the activate command again.

If you see a permissions error, you are likely running outside the venv. Never use sudo pip install. Activate the venv first.

Help me troubleshoot my pip install failure for this project's dependencies.

Open the project in VS Code and verify the environment
  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Visual Studio Code and press Enter to open it.
  • In VS Code, click File in the top menu bar.
  • Click Open Folder.
  • Select the research-extraction-api folder on your Desktop, then click Open.

Now verify that Python and your installed packages are working correctly inside the project.

  • Open the integrated terminal in VS Code by pressing Ctrl+` (backtick key, below Escape).
  • Make sure your virtual environment is active. If you do not see (venv) in the prompt, activate it again.
  • Confirm your Python version by running:
python --version

You should see Python 3.13.x in the output.

  • Run a quick import test to confirm all packages installed correctly:
python -c "import fastapi; import uvicorn; import pynetdicom; import pydicom; import smbclient; print('All packages imported successfully')"

You should see All packages imported successfully printed with no errors.

Seeing a ModuleNotFoundError?

This means your terminal is not using the virtual environment where you installed the packages. Check that (venv) appears in your prompt. If it does not, activate the environment and try again.

Also confirm that which python (macOS) or where python (Windows) points to the venv folder inside your project.

Help me fix my ModuleNotFoundError when importing packages in my research-extraction-api project.

✔️ Awesome, I've got everything!

Great. Double check you have saved your requirements.txt file.

ⓧ I'd like to double check the full code

Here is what your requirements.txt should contain:

fastapi==0.136.1
uvicorn[standard]==0.34.0
pynetdicom==3.0.4
pydicom==3.0.1
smbprotocol==1.16.1
rfc5424-logging-handler==1.4.3

Your development environment is ready. Next up, you will configure a local test PACS using dcmqrscp so your DICOM adapter has something to talk to.

Configure dcmqrscp as a Local Test PACS

Your API needs a source to pull DICOM imaging data from. In a hospital, that source is a PACS (Picture Archiving and Communication System). For local development, you can simulate one using dcmqrscp, a lightweight DICOM archive included with DCMTK.

In this step, you will configure dcmqrscp to listen on port 11112, load it with sample DICOM files, and verify your Python client can reach it over the network.

In this step, get ready to:
  • Create a dcmqrscp configuration file that defines your test PACS.
  • Upload sample DICOM files into the archive using storescu.
  • Start dcmqrscp and verify connectivity with echoscu.
Create the dcmqrscp configuration file

The dcmqrscp tool reads a configuration file that defines network settings, known peers, and storage areas. You need to create this file so dcmqrscp knows which port to listen on, which AE titles are allowed to connect, and where to store received images.

  • In VS Code, create a new file called dcmqrscp.cfg in the root of your research-extraction-api folder.
  • Paste the following configuration into the file:
NetworkTCPPort  = 11112
MaxPDUSize      = 16384
MaxAssociations = 16

HostTable BEGIN
research_scu = (RESEARCH_SCU, localhost, 11113)
HostTable END

VendorTable BEGIN
VendorTable END

AETable BEGIN
DCMQRSCP   ./dicom_store   RW (200, 1024mb)   ANY
AETable END

What does this configuration do?

The file has four sections that dcmqrscp reads on startup:

  • Global parameters set the TCP port (11112), maximum PDU size, and maximum concurrent associations.
  • HostTable registers your Python client (RESEARCH_SCU) so dcmqrscp knows who it is.
  • VendorTable is required by the config format but can be empty for testing.
  • AETable defines the storage area. DCMQRSCP is the AE title, ./dicom_store is where files are saved, RW allows read/write, and ANY permits connections from any calling AE title.
  • Save dcmqrscp.cfg.
  • Confirm the file exists by running:
ls dcmqrscp.cfg

You should see dcmqrscp.cfg listed in the output.

File not found?

Make sure your terminal is in the research-extraction-api directory. Run pwd to check your current location.

Help me find my dcmqrscp.cfg file

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before moving on.

ⓧ I'd like to double check the full code

NetworkTCPPort  = 11112
MaxPDUSize      = 16384
MaxAssociations = 16

HostTable BEGIN
research_scu = (RESEARCH_SCU, localhost, 11113)
HostTable END

VendorTable BEGIN
VendorTable END

AETable BEGIN
DCMQRSCP   ./dicom_store   RW (200, 1024mb)   ANY
AETable END
Create the storage directory and upload sample DICOM files

The AETable references ./dicom_store as the storage area. You need to create this directory before dcmqrscp can start. Then you will generate a test DICOM file using pydicom and upload it with storescu.

  • Create the storage directory by running:
mkdir dicom_store
  • Confirm it exists by running:
ls

You should see dicom_store listed alongside dcmqrscp.cfg and requirements.txt.

  • Generate a test DICOM file using pydicom by running this command:
python -c "
from pydicom.dataset import Dataset, FileDataset, FileMetaDataset
from pydicom.uid import ExplicitVRLittleEndian, generate_uid

file_meta = FileMetaDataset()
file_meta.MediaStorageSOPClassUID = '1.2.840.10008.5.1.4.1.1.2'
file_meta.MediaStorageSOPInstanceUID = generate_uid()
file_meta.TransferSyntaxUID = ExplicitVRLittleEndian
file_meta.ImplementationClassUID = '1.2.826.0.1.3680043.8.498.1'

ds = FileDataset('test_ct.dcm', Dataset(), file_meta=file_meta, preamble=b'\x00'*128)
ds.SOPClassUID = '1.2.840.10008.5.1.4.1.1.2'
ds.SOPInstanceUID = file_meta.MediaStorageSOPInstanceUID
ds.StudyInstanceUID = '1.2.826.0.1.3680043.8.1055.1.20111103111148288.98361414.79379639'
ds.SeriesInstanceUID = generate_uid()
ds.PatientName = 'Test^Patient'
ds.PatientID = 'TEST001'
ds.Modality = 'CT'
ds.save_as('test_ct.dcm', enforce_file_format=True)
print('Created test_ct.dcm')
"

What does this script do?

This creates a minimal valid DICOM file with CT Image metadata. The StudyInstanceUID is a fixed value you will use later to test retrieval from your API.

The enforce_file_format=True flag ensures the file includes the required 128-byte preamble and DICM prefix that DICOM tools expect.

You should see Created test_ct.dcm printed in your terminal.

  • Now start dcmqrscp temporarily so you can upload the file. Open a second terminal in VS Code by pressing Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows), then typing Terminal: Create New Terminal and pressing Enter.
  • In the new terminal, activate your virtual environment and start dcmqrscp by running:
dcmqrscp --config dcmqrscp.cfg

The terminal will appear to hang. This is normal. dcmqrscp is now listening for DICOM connections and will not exit on its own.

  • Switch back to your first terminal and upload the test file by running:
storescu --aetitle RESEARCH_SCU --call DCMQRSCP localhost 11112 test_ct.dcm

What does storescu do?

The storescu command sends a DICOM file to dcmqrscp using the Storage Service Class. The --aetitle flag sets your calling AE title, and --call sets the AE title of the receiving PACS. The file gets stored in the ./dicom_store directory.

  • Verify the file was stored by running:
ls dicom_store/

You should see at least one .dcm file listed in the output.

storescu says connection refused?

Make sure dcmqrscp is running in your other terminal. Check that you are using port 11112 and the AE title DCMQRSCP (case-sensitive). Also confirm you are running storescu from the research-extraction-api directory where test_ct.dcm exists.

Help me troubleshoot this storescu error

Verify connectivity with echoscu

A DICOM Echo (C-ECHO) is the equivalent of a network ping for DICOM systems. It confirms that your client can establish a DICOM association with the PACS.

  • With dcmqrscp still running in your second terminal, verify connectivity by running:
echoscu --aetitle RESEARCH_SCU --call DCMQRSCP localhost 11112

If the command completes silently with no error output, the connection is working. A successful echo returns exit code 0 and prints nothing.

What if I want to see more detail?

Add the -v flag to see verbose association negotiation output. Look for Association Accepted and Releasing Association in the output to confirm success.

Seeing an association rejected error?

Double-check that the --call value matches the AE title in your AETable (DCMQRSCP). AE titles are case-sensitive. Also ensure the --aetitle matches what you registered in the HostTable (RESEARCH_SCU).

Help me debug this association rejected error

Keep dcmqrscp running

Leave dcmqrscp running in its terminal for the rest of the project. Your DICOM adapter will connect to it when you test extraction requests in later steps.

Your local test PACS is up and loaded with sample data. Next up, you will build the FastAPI skeleton that accepts extraction requests and dispatches them to source adapters.

Build the FastAPI Skeleton with Authentication

Your test PACS is running and accepting connections. Now it's time to build the API service that will orchestrate data extraction from that PACS and other sources.

FastAPI gives you a high-performance Python web framework with automatic interactive documentation. In this step, you'll create the core application with request models, API key authentication, and two endpoints that handle extraction jobs.

In this step, get ready to:
  • Create main.py with the FastAPI app, Pydantic data models, and authentication.
  • Add the extraction endpoints that dispatch background tasks.
  • Run the server and test authentication in Swagger UI.
Set up main.py with the app, authentication, and models

Every FastAPI application starts with an app instance and data models that define the shape of requests and responses. You also need an authentication layer to prevent unauthorized access to your extraction service.

  • Create a new file called main.py in your research-extraction-api folder by running:
touch main.py
  • Confirm the file exists by running:
ls

You should see main.py listed alongside requirements.txt and dcmqrscp.cfg.

  • Open main.py in VS Code and paste the following code:
import uuid
from fastapi import FastAPI, BackgroundTasks, Depends, HTTPException, Security
from fastapi.security import APIKeyHeader
from pydantic import BaseModel

app = FastAPI(title="Research Data Extraction API", version="1.0.0")

# --- Authentication ---
API_KEY_HEADER = APIKeyHeader(name="X-API-Key")
VALID_API_KEYS = ["research-key-001", "research-key-002"]


def verify_api_key(api_key: str = Security(API_KEY_HEADER)):
    if api_key not in VALID_API_KEYS:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return api_key


# --- Job tracking ---
jobs: dict = {}


# --- Request/Response models ---
class ExtractionRequest(BaseModel):
    source_type: str  # "dicom" or "smb"
    identifiers: list[str]  # UIDs for DICOM, folder paths for SMB
    destination: str  # local staging path


class ExtractionResponse(BaseModel):
    job_id: str
    status: str


class JobStatus(BaseModel):
    job_id: str
    status: str
    files_extracted: list[str]

What does this code do?

This sets up the foundation of your extraction API.

  • APIKeyHeader tells FastAPI to look for an X-API-Key header on every protected request.
  • verify_api_key is a dependency function that FastAPI calls automatically before your endpoint runs. If the key is invalid, it short-circuits with a 401 response.
  • The three Pydantic models define the exact JSON shape that the API accepts and returns. FastAPI uses these to validate input and generate documentation.

Why API keys instead of OAuth2?

API keys are simple, stateless, and appropriate for service-to-service calls in a research environment. OAuth2 or JWT adds complexity that isn't needed for this scope. The Secret Mission adds LDAP authentication for production deployment where user-level identity matters.

  • Save main.py.

Seeing a red underline on imports?

Make sure your virtual environment is activated and selected as the Python interpreter in VS Code. Click the Python version in the bottom status bar and choose the interpreter from your project's virtual environment.

If FastAPI isn't found, re-run pip install fastapi with your virtual environment active.

Why are my FastAPI imports not resolving in VS Code?

Add the extraction endpoints

Your models define the data contract. Now you need the endpoints that accept requests and dispatch extraction work to a background task.

  • In main.py, add the following code below your JobStatus model:
# --- Background task (stub) ---
def run_extraction(job_id: str, request: ExtractionRequest, requester: str):
    jobs[job_id]["status"] = "completed"


# --- Endpoints ---
@app.post("/extract", response_model=ExtractionResponse)
def extract_data(
    request: ExtractionRequest,
    background_tasks: BackgroundTasks,
    api_key: str = Depends(verify_api_key),
):
    job_id = str(uuid.uuid4())
    jobs[job_id] = {"status": "queued", "files_extracted": []}
    background_tasks.add_task(run_extraction, job_id, request, api_key)
    return ExtractionResponse(job_id=job_id, status="queued")


@app.get("/status/{job_id}", response_model=JobStatus)
def get_status(job_id: str, api_key: str = Depends(verify_api_key)):
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")
    job = jobs[job_id]
    return JobStatus(
        job_id=job_id,
        status=job["status"],
        files_extracted=job.get("files_extracted", []),
    )

What does this code do?

  • run_extraction is a stub that immediately marks a job as completed. You'll replace this with real DICOM and SMB logic in later steps.
  • POST /extract creates a unique job ID, stores it in the jobs dictionary, and dispatches the extraction to a BackgroundTasks runner. The response returns immediately with the job ID so the client doesn't block.
  • GET /status/{job_id} lets the client poll for completion. Both endpoints require a valid API key via Depends(verify_api_key).
  • Save main.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here's the complete main.py file for comparison:

import uuid
from fastapi import FastAPI, BackgroundTasks, Depends, HTTPException, Security
from fastapi.security import APIKeyHeader
from pydantic import BaseModel

app = FastAPI(title="Research Data Extraction API", version="1.0.0")

# --- Authentication ---
API_KEY_HEADER = APIKeyHeader(name="X-API-Key")
VALID_API_KEYS = ["research-key-001", "research-key-002"]


def verify_api_key(api_key: str = Security(API_KEY_HEADER)):
    if api_key not in VALID_API_KEYS:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return api_key


# --- Job tracking ---
jobs: dict = {}


# --- Request/Response models ---
class ExtractionRequest(BaseModel):
    source_type: str  # "dicom" or "smb"
    identifiers: list[str]  # UIDs for DICOM, folder paths for SMB
    destination: str  # local staging path


class ExtractionResponse(BaseModel):
    job_id: str
    status: str


class JobStatus(BaseModel):
    job_id: str
    status: str
    files_extracted: list[str]


# --- Background task (stub) ---
def run_extraction(job_id: str, request: ExtractionRequest, requester: str):
    jobs[job_id]["status"] = "completed"


# --- Endpoints ---
@app.post("/extract", response_model=ExtractionResponse)
def extract_data(
    request: ExtractionRequest,
    background_tasks: BackgroundTasks,
    api_key: str = Depends(verify_api_key),
):
    job_id = str(uuid.uuid4())
    jobs[job_id] = {"status": "queued", "files_extracted": []}
    background_tasks.add_task(run_extraction, job_id, request, api_key)
    return ExtractionResponse(job_id=job_id, status="queued")


@app.get("/status/{job_id}", response_model=JobStatus)
def get_status(job_id: str, api_key: str = Depends(verify_api_key)):
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")
    job = jobs[job_id]
    return JobStatus(
        job_id=job_id,
        status=job["status"],
        files_extracted=job.get("files_extracted", []),
    )
Run the server and test in Swagger UI

FastAPI automatically generates interactive API documentation from your Pydantic models and endpoint definitions. You can use this to test your endpoints without writing any client code.

  • Start the development server by running:
uvicorn main:app --reload

What does this command do?

  • uvicorn is the ASGI server that runs your FastAPI application.
  • main:app tells uvicorn to look for the app object in main.py.
  • --reload watches for file changes and restarts the server automatically during development.

You should see output showing the server started on http://127.0.0.1:8000.

  • Open your browser and navigate to:
http://localhost:8000/docs

You should see the Swagger UI with your two endpoints listed: POST /extract and GET /status/{job_id}. There's also an Authorize button at the top for entering your API key.

  • Click the Authorize button at the top of the page.
  • Enter research-key-001 in the value field.
  • Click Authorize, then Close.
  • Expand the POST /extract endpoint and click Try it out.
  • Replace the request body with this JSON:
{
  "source_type": "dicom",
  "identifiers": ["1.2.3.4.5"],
  "destination": "./staging"
}
  • Click Execute.

You should see a 200 response with a job_id (a UUID) and status: "queued".

  • Now test that authentication rejects invalid keys. Click Authorize again.
  • Click Logout to clear your current key.
  • Enter wrong-key-999 as the value and click Authorize.
  • Try executing the POST /extract endpoint again with the same request body.

You should see a 401 response with "detail": "Invalid API key". Your authentication is working correctly.

Server won't start?

  • If you see ModuleNotFoundError: No module named 'fastapi', make sure your virtual environment is activated before running uvicorn.
  • If you see a SyntaxError, double check that your main.py matches the full code in the tab above.
  • If port 8000 is already in use, stop any other process using that port or run with uvicorn main:app --reload --port 8001.

Why won't my uvicorn server start?

Your API skeleton is live and protected. Next up, you'll build the ATNA audit logger so every extraction event gets recorded in a compliance-ready format.

Implement the ATNA Audit Logger

Your API can now authenticate requests and track job status. The next piece of the puzzle is compliance.

In healthcare environments, every time someone accesses patient data, that access must be recorded. IHE ATNA (Audit Trail and Node Authentication) defines the standard format for these records. In this step, you'll build a module that generates compliant audit messages every time data is extracted.

In this step, get ready to:
  • Configure the ATNA audit logger module with file-based output.
  • Build a DICOM PS3.15 A.5 compliant audit message generator.
  • Create the data export event logging function and verify it writes audit XML.
Set up the audit module and logger

The audit logger writes structured XML messages to a log file. In production, you would swap the file handler for an RFC 5424 syslog handler that transmits over TLS to an Audit Record Repository.

  • Create a new file called audit.py in your research-extraction-api folder (the same level as main.py).
  • Add the imports and logger configuration by pasting this code:
import logging
import socket
from datetime import datetime, timezone
from xml.etree.ElementTree import Element, SubElement, tostring

# Set up a file-based logger for ATNA audit messages
audit_logger = logging.getLogger("atna.audit")
audit_logger.setLevel(logging.INFO)
_handler = logging.FileHandler("atna_audit.log")
_handler.setFormatter(logging.Formatter("%(message)s"))
audit_logger.addHandler(_handler)

# Constants identifying this service in audit messages
AUDIT_SOURCE_ID = "ResearchExtractionAPI"
HOSTNAME = socket.gethostname()

What does this code do?

The logger is configured with a plain %(message)s format because the message itself is structured XML. No additional formatting is needed.

The AUDIT_SOURCE_ID identifies your service in every audit record. HOSTNAME captures the machine name for the network access point fields.

Why log to a file instead of syslog?

For local development, writing to a file lets you inspect audit messages directly. In production, you would replace FileHandler with rfc5424-logging-handler's Rfc5424SysLogHandler to transmit over TLS to your hospital's Audit Record Repository.

  • Save audit.py.
Build the ATNA audit message generator

ATNA audit messages follow the DICOM PS3.15 A.5 schema. Each message contains event metadata, participant information, and object identifiers that describe exactly who accessed what data and when.

  • In audit.py, add the build_audit_message function below your constants by pasting this code:
def build_audit_message(
    requester: str,
    source_type: str,
    identifiers: list[str],
    outcome: int,
) -> str:
    # Root element for the ATNA audit XML
    root = Element("AuditMessage")

    # EventIdentification - what happened and when
    event = SubElement(root, "EventIdentification")
    event.set("EventActionCode", "R")
    event.set("EventDateTime", datetime.now(timezone.utc).isoformat())
    event.set("EventOutcomeIndicator", str(outcome))

    # EventID - categorize as an Export event (DCM code 110106)
    event_id = SubElement(event, "EventID")
    event_id.set("csd-code", "110106")
    event_id.set("codeSystemName", "DCM")
    event_id.set("displayName", "Export")

    # EventTypeCode - IHE transaction type
    event_type = SubElement(event, "EventTypeCode")
    event_type.set("csd-code", "ITI-" + ("41" if source_type == "dicom" else "43"))
    event_type.set("codeSystemName", "IHE Transactions")
    event_type.set("displayName", f"Data Export ({source_type})")

    # ActiveParticipant - the requester (source of the request)
    source_participant = SubElement(root, "ActiveParticipant")
    source_participant.set("UserID", requester)
    source_participant.set("UserIsRequestor", "true")
    source_participant.set("NetworkAccessPointID", HOSTNAME)
    source_participant.set("NetworkAccessPointTypeCode", "1")

    role_source = SubElement(source_participant, "RoleIDCode")
    role_source.set("csd-code", "110153")
    role_source.set("codeSystemName", "DCM")
    role_source.set("displayName", "Source Role ID")

    # ActiveParticipant - the destination (this service)
    dest_participant = SubElement(root, "ActiveParticipant")
    dest_participant.set("UserID", AUDIT_SOURCE_ID)
    dest_participant.set("UserIsRequestor", "false")
    dest_participant.set("NetworkAccessPointID", HOSTNAME)
    dest_participant.set("NetworkAccessPointTypeCode", "1")

    role_dest = SubElement(dest_participant, "RoleIDCode")
    role_dest.set("csd-code", "110152")
    role_dest.set("codeSystemName", "DCM")
    role_dest.set("displayName", "Destination Role ID")

    # AuditSourceIdentification - which system generated this audit
    audit_source = SubElement(root, "AuditSourceIdentification")
    audit_source.set("AuditSourceID", AUDIT_SOURCE_ID)

    # ParticipantObjectIdentification - one per data object accessed
    for identifier in identifiers:
        participant_obj = SubElement(root, "ParticipantObjectIdentification")
        participant_obj.set("ParticipantObjectID", identifier)
        participant_obj.set("ParticipantObjectTypeCode", "2")
        participant_obj.set("ParticipantObjectTypeCodeRole", "3")

        obj_id_type = SubElement(participant_obj, "ParticipantObjectIDTypeCode")
        if source_type == "dicom":
            obj_id_type.set("csd-code", "110180")
            obj_id_type.set("codeSystemName", "DCM")
            obj_id_type.set("displayName", "Study Instance UID")
        else:
            obj_id_type.set("csd-code", "12")
            obj_id_type.set("codeSystemName", "RFC-3881")
            obj_id_type.set("displayName", "URI")

    return tostring(root, encoding="unicode", xml_declaration=True)

What does this function build?

The function constructs an XML document with four required ATNA sections:

  • EventIdentification records the action type (R for Read), timestamp, and outcome (0 = success, 8 = failure).
  • ActiveParticipant identifies both the requester (the API key holder) and the destination (this service). Role codes 110153 and 110152 are DCM standard codes for Source and Destination.
  • AuditSourceIdentification tags which system generated this audit record.
  • ParticipantObjectIdentification lists each data object accessed. DICOM objects use code 110180 (Study Instance UID). SMB files use code 12 (URI) from RFC-3881.

Why hand-craft the XML?

No Python library exists for generating ATNA audit messages (Java has IPF commons-audit, but Python does not). The XML schema is straightforward enough to build directly with xml.etree.ElementTree.

  • Save audit.py.

Seeing a syntax error?

Check that the f-string on the displayName line uses regular double quotes around the outer string and regular quotes inside. Mixed quote types cause parsing errors.

Also verify the function signature ends with ) -> str: on its own line.

Help me fix a syntax error in my audit.py file

Add the export event function and verify

The log_data_export_event function is the public interface that the rest of your application will call. It wraps the message builder and sends the result to the audit logger.

  • At the bottom of audit.py, add the log_data_export_event function by pasting this code:
def log_data_export_event(
    requester: str,
    source_type: str,
    identifiers: list[str],
    outcome: int = 0,
):
    # Build the ATNA XML and write it to the audit log
    message = build_audit_message(requester, source_type, identifiers, outcome)
    audit_logger.info(message)

What does this code do?

This thin wrapper provides a clean API for the rest of your application. The outcome parameter defaults to 0 (success). When an extraction fails, the caller passes 8 (serious failure) so the audit trail reflects the error.

  • Save audit.py.
  • Open a new terminal in VS Code by pressing Ctrl+` (macOS and Windows).
  • Test the audit logger by running this command:
python -c "from audit import log_data_export_event; log_data_export_event(requester='test-key', source_type='dicom', identifiers=['1.2.3.4.5'])"
  • Verify the audit log was created by running:
cat atna_audit.log

You should see an XML document starting with <?xml version='1.0' encoding='us-ascii'?> followed by an <AuditMessage> root element containing EventIdentification, ActiveParticipant, and ParticipantObjectIdentification elements.

Don't see the XML output?

Make sure your terminal is in the research-extraction-api directory where audit.py lives. The from audit import statement will fail with ModuleNotFoundError if you are in the wrong directory.

Also check that your virtual environment is still activated. You should see the environment name in your terminal prompt.

Help me debug why my audit.py import is failing

✔️ Awesome, I've got everything!

Great. Double check you've saved audit.py and move on to the next step.

ⓧ I'd like to double check the full code

Here is the complete audit.py file:

import logging
import socket
from datetime import datetime, timezone
from xml.etree.ElementTree import Element, SubElement, tostring

audit_logger = logging.getLogger("atna.audit")
audit_logger.setLevel(logging.INFO)
_handler = logging.FileHandler("atna_audit.log")
_handler.setFormatter(logging.Formatter("%(message)s"))
audit_logger.addHandler(_handler)

AUDIT_SOURCE_ID = "ResearchExtractionAPI"
HOSTNAME = socket.gethostname()


def build_audit_message(
    requester: str,
    source_type: str,
    identifiers: list[str],
    outcome: int,
) -> str:
    root = Element("AuditMessage")

    event = SubElement(root, "EventIdentification")
    event.set("EventActionCode", "R")
    event.set("EventDateTime", datetime.now(timezone.utc).isoformat())
    event.set("EventOutcomeIndicator", str(outcome))

    event_id = SubElement(event, "EventID")
    event_id.set("csd-code", "110106")
    event_id.set("codeSystemName", "DCM")
    event_id.set("displayName", "Export")

    event_type = SubElement(event, "EventTypeCode")
    event_type.set("csd-code", "ITI-" + ("41" if source_type == "dicom" else "43"))
    event_type.set("codeSystemName", "IHE Transactions")
    event_type.set("displayName", f"Data Export ({source_type})")

    source_participant = SubElement(root, "ActiveParticipant")
    source_participant.set("UserID", requester)
    source_participant.set("UserIsRequestor", "true")
    source_participant.set("NetworkAccessPointID", HOSTNAME)
    source_participant.set("NetworkAccessPointTypeCode", "1")

    role_source = SubElement(source_participant, "RoleIDCode")
    role_source.set("csd-code", "110153")
    role_source.set("codeSystemName", "DCM")
    role_source.set("displayName", "Source Role ID")

    dest_participant = SubElement(root, "ActiveParticipant")
    dest_participant.set("UserID", AUDIT_SOURCE_ID)
    dest_participant.set("UserIsRequestor", "false")
    dest_participant.set("NetworkAccessPointID", HOSTNAME)
    dest_participant.set("NetworkAccessPointTypeCode", "1")

    role_dest = SubElement(dest_participant, "RoleIDCode")
    role_dest.set("csd-code", "110152")
    role_dest.set("codeSystemName", "DCM")
    role_dest.set("displayName", "Destination Role ID")

    audit_source = SubElement(root, "AuditSourceIdentification")
    audit_source.set("AuditSourceID", AUDIT_SOURCE_ID)

    for identifier in identifiers:
        participant_obj = SubElement(root, "ParticipantObjectIdentification")
        participant_obj.set("ParticipantObjectID", identifier)
        participant_obj.set("ParticipantObjectTypeCode", "2")
        participant_obj.set("ParticipantObjectTypeCodeRole", "3")

        obj_id_type = SubElement(participant_obj, "ParticipantObjectIDTypeCode")
        if source_type == "dicom":
            obj_id_type.set("csd-code", "110180")
            obj_id_type.set("codeSystemName", "DCM")
            obj_id_type.set("displayName", "Study Instance UID")
        else:
            obj_id_type.set("csd-code", "12")
            obj_id_type.set("codeSystemName", "RFC-3881")
            obj_id_type.set("displayName", "URI")

    return tostring(root, encoding="unicode", xml_declaration=True)


def log_data_export_event(
    requester: str,
    source_type: str,
    identifiers: list[str],
    outcome: int = 0,
):
    message = build_audit_message(requester, source_type, identifiers, outcome)
    audit_logger.info(message)

Your audit trail is ready. Next up, you'll build the DICOM adapter that actually retrieves imaging data from your local PACS and triggers these audit messages on every extraction.

Implement the DICOM Adapter

Your API skeleton accepts requests and authenticates callers, but the background task is still a stub. The real goal is to reach into a PACS and pull imaging data back to a local staging directory.

In this step, you will build a DICOM adapter that uses pynetdicom's C-GET operation to retrieve studies from your local dcmqrscp. You will also replace the stub run_extraction function with real dispatch logic.

In this step, get ready to:
  • Create the DICOM adapter module with C-GET retrieval logic.
  • Wire the adapter into the extract endpoint with audit logging.
  • Test a full DICOM extraction against your local test PACS.
Create the DICOM adapter module

The adapter pattern isolates each data source behind a consistent interface. Your DICOM adapter will accept a list of Study Instance UIDs and a destination path, then use C-GET to pull every instance in those studies from the PACS.

Why C-GET instead of C-MOVE?

C-GET retrieves files over the same association. You do not need to register a separate Storage SCP with the PACS or configure firewall rules for an inbound connection.

C-MOVE requires the PACS to initiate a new connection back to your machine, which adds operational complexity. For a self-contained research tool, C-GET is simpler and more portable.

  • Create the adapters/ directory inside your research-extraction-api project folder by running these commands:
mkdir adapters
touch adapters/__init__.py
  • Confirm the files exist by running:
ls adapters/

You should see __init__.py in the output.

  • Create a new file called adapters/dicom_adapter.py in VS Code and paste the following code:
import os
from pydicom.dataset import Dataset
from pynetdicom import AE, evt, build_role, StoragePresentationContexts
from pynetdicom.sop_class import (
    PatientRootQueryRetrieveInformationModelGet,
)

# Configuration - replace with your PACS details
PACS_HOST = "127.0.0.1"
PACS_PORT = 11112
PACS_AE_TITLE = "DCMQRSCP"
LOCAL_AE_TITLE = "RESEARCH_SCU"


def handle_store(event, destination):
    """Handle incoming C-STORE from the PACS during C-GET."""
    ds = event.dataset
    ds.file_meta = event.file_meta
    filename = f"{ds.SOPInstanceUID}.dcm"
    filepath = os.path.join(destination, filename)
    ds.save_as(filepath, enforce_file_format=True)
    return 0x0000


def retrieve_dicom(identifiers: list[str], destination: str) -> list[str]:
    """Retrieve DICOM instances from PACS using C-GET."""
    os.makedirs(destination, exist_ok=True)
    extracted_files = []

    ae = AE(ae_title=LOCAL_AE_TITLE)
    ae.add_requested_context(PatientRootQueryRetrieveInformationModelGet)

    for cx in StoragePresentationContexts:
        ae.add_requested_context(cx.abstract_syntax)

    ext_neg = []
    for cx in StoragePresentationContexts:
        ext_neg.append(build_role(cx.abstract_syntax, scp_role=True))

    handlers = [
        (evt.EVT_C_STORE, handle_store, [destination]),
    ]

    for uid in identifiers:
        ds = Dataset()
        ds.QueryRetrieveLevel = "STUDY"
        ds.StudyInstanceUID = uid

        assoc = ae.associate(
            PACS_HOST,
            PACS_PORT,
            ae_title=PACS_AE_TITLE,
            ext_neg=ext_neg,
            evt_handlers=handlers,
        )

        if assoc.is_established:
            responses = assoc.send_c_get(
                ds, PatientRootQueryRetrieveInformationModelGet
            )
            for status, identifier in responses:
                if status and status.Status in [0xFF00, 0xFF01]:
                    pass
            assoc.release()

    extracted_files = [
        f for f in os.listdir(destination) if f.endswith(".dcm")
    ]
    return extracted_files

What does this code do?

The handle_store function is a callback. When the PACS sends back an image during C-GET, pynetdicom fires this handler so you can save the file to disk.

The retrieve_dicom function sets up the association with role negotiation. It tells the PACS that your application can act as a Storage SCP (receiver) by passing build_role(cx.abstract_syntax, scp_role=True) for every storage context. Without this, the PACS will refuse to send images back.

For each Study Instance UID, the function builds a query dataset at the STUDY level and calls send_c_get. The PACS responds with pending statuses (0xFF00, 0xFF01) while streaming files back.

  • Save adapters/dicom_adapter.py.

You will not see a visible change yet. The next substep wires this adapter into your API endpoint.

Seeing an import error in VS Code?

Make sure your VS Code Python interpreter is set to the virtual environment inside research-extraction-api. Press Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows), type Python: Select Interpreter, and choose the one inside your venv folder.

help me fix pynetdicom import errors in my DICOM adapter

Wire the adapter into the extract endpoint

The adapter is ready, but your main.py still uses a stub that immediately marks jobs as completed. You need to import the adapter, replace the stub with real dispatch logic, and log every extraction event through the ATNA audit logger.

  • In main.py, find these lines at the top of the file:
import uuid
from fastapi import FastAPI, BackgroundTasks, Depends, HTTPException, Security
from fastapi.security import APIKeyHeader
from pydantic import BaseModel
  • Add two new import lines after from pydantic import BaseModel, so the import block looks like this:
import uuid
from fastapi import FastAPI, BackgroundTasks, Depends, HTTPException, Security
from fastapi.security import APIKeyHeader
from pydantic import BaseModel
from adapters.dicom_adapter import retrieve_dicom
from audit import log_data_export_event

What do these imports add?

The retrieve_dicom import pulls in the C-GET function you just created. The log_data_export_event import brings in the ATNA audit logger from the previous step so every extraction is recorded.

  • Now find the stub run_extraction function:
# --- Background task (stub) ---
def run_extraction(job_id: str, request: ExtractionRequest, requester: str):
    jobs[job_id]["status"] = "completed"
  • Replace the entire stub function (including the comment above it) with this implementation:
# --- Background task ---
def run_extraction(job_id: str, request: ExtractionRequest, requester: str):
    try:
        jobs[job_id]["status"] = "running"
        extracted_files = []

        if request.source_type == "dicom":
            extracted_files = retrieve_dicom(
                identifiers=request.identifiers,
                destination=request.destination,
            )
        else:
            jobs[job_id]["status"] = "failed"
            jobs[job_id]["error"] = f"Unknown source_type: {request.source_type}"
            log_data_export_event(
                requester=requester,
                source_type=request.source_type,
                identifiers=request.identifiers,
                outcome=8,
            )
            return

        jobs[job_id]["status"] = "completed"
        jobs[job_id]["files_extracted"] = extracted_files

        log_data_export_event(
            requester=requester,
            source_type=request.source_type,
            identifiers=request.identifiers,
            outcome=0,
        )
    except Exception as e:
        jobs[job_id]["status"] = "failed"
        jobs[job_id]["error"] = str(e)
        log_data_export_event(
            requester=requester,
            source_type=request.source_type,
            identifiers=request.identifiers,
            outcome=8,
        )

What does this code do?

The function wraps everything in a try/except so that crashes during retrieval are caught and logged.

When source_type is "dicom", it calls retrieve_dicom. Any other type fails the job. On success, the ATNA logger records outcome 0 (success). On failure, it records outcome 8 (serious failure).

  • Save main.py.

If uvicorn is still running with --reload, it will pick up the changes automatically.

Seeing a ModuleNotFoundError for adapters?

Confirm that adapters/__init__.py exists. Without this file, Python does not recognize adapters/ as a package.

Also confirm you are running uvicorn from the research-extraction-api folder (not from inside the adapters/ subfolder).

help me fix the ModuleNotFoundError for my adapters package

✔️ Awesome, I've got everything!

Great. Double check you have saved both adapters/dicom_adapter.py and main.py.

ⓧ I'd like to double check the full code

Here is the full adapters/dicom_adapter.py:

import os
from pydicom.dataset import Dataset
from pynetdicom import AE, evt, build_role, StoragePresentationContexts
from pynetdicom.sop_class import (
    PatientRootQueryRetrieveInformationModelGet,
)

# Configuration - replace with your PACS details
PACS_HOST = "127.0.0.1"
PACS_PORT = 11112
PACS_AE_TITLE = "DCMQRSCP"
LOCAL_AE_TITLE = "RESEARCH_SCU"


def handle_store(event, destination):
    """Handle incoming C-STORE from the PACS during C-GET."""
    ds = event.dataset
    ds.file_meta = event.file_meta
    filename = f"{ds.SOPInstanceUID}.dcm"
    filepath = os.path.join(destination, filename)
    ds.save_as(filepath, enforce_file_format=True)
    return 0x0000


def retrieve_dicom(identifiers: list[str], destination: str) -> list[str]:
    """Retrieve DICOM instances from PACS using C-GET."""
    os.makedirs(destination, exist_ok=True)
    extracted_files = []

    ae = AE(ae_title=LOCAL_AE_TITLE)
    ae.add_requested_context(PatientRootQueryRetrieveInformationModelGet)

    for cx in StoragePresentationContexts:
        ae.add_requested_context(cx.abstract_syntax)

    ext_neg = []
    for cx in StoragePresentationContexts:
        ext_neg.append(build_role(cx.abstract_syntax, scp_role=True))

    handlers = [
        (evt.EVT_C_STORE, handle_store, [destination]),
    ]

    for uid in identifiers:
        ds = Dataset()
        ds.QueryRetrieveLevel = "STUDY"
        ds.StudyInstanceUID = uid

        assoc = ae.associate(
            PACS_HOST,
            PACS_PORT,
            ae_title=PACS_AE_TITLE,
            ext_neg=ext_neg,
            evt_handlers=handlers,
        )

        if assoc.is_established:
            responses = assoc.send_c_get(
                ds, PatientRootQueryRetrieveInformationModelGet
            )
            for status, identifier in responses:
                if status and status.Status in [0xFF00, 0xFF01]:
                    pass
            assoc.release()

    extracted_files = [
        f for f in os.listdir(destination) if f.endswith(".dcm")
    ]
    return extracted_files

Here is the full main.py:

import uuid
from fastapi import FastAPI, BackgroundTasks, Depends, HTTPException, Security
from fastapi.security import APIKeyHeader
from pydantic import BaseModel
from adapters.dicom_adapter import retrieve_dicom
from audit import log_data_export_event

app = FastAPI(title="Research Data Extraction API", version="1.0.0")

# --- Authentication ---
API_KEY_HEADER = APIKeyHeader(name="X-API-Key")
VALID_API_KEYS = ["research-key-001", "research-key-002"]


def verify_api_key(api_key: str = Security(API_KEY_HEADER)):
    if api_key not in VALID_API_KEYS:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return api_key


# --- Job tracking ---
jobs: dict = {}


# --- Request/Response models ---
class ExtractionRequest(BaseModel):
    source_type: str
    identifiers: list[str]
    destination: str


class ExtractionResponse(BaseModel):
    job_id: str
    status: str


class JobStatus(BaseModel):
    job_id: str
    status: str
    files_extracted: list[str]


# --- Background task ---
def run_extraction(job_id: str, request: ExtractionRequest, requester: str):
    try:
        jobs[job_id]["status"] = "running"
        extracted_files = []

        if request.source_type == "dicom":
            extracted_files = retrieve_dicom(
                identifiers=request.identifiers,
                destination=request.destination,
            )
        else:
            jobs[job_id]["status"] = "failed"
            jobs[job_id]["error"] = f"Unknown source_type: {request.source_type}"
            log_data_export_event(
                requester=requester,
                source_type=request.source_type,
                identifiers=request.identifiers,
                outcome=8,
            )
            return

        jobs[job_id]["status"] = "completed"
        jobs[job_id]["files_extracted"] = extracted_files

        log_data_export_event(
            requester=requester,
            source_type=request.source_type,
            identifiers=request.identifiers,
            outcome=0,
        )
    except Exception as e:
        jobs[job_id]["status"] = "failed"
        jobs[job_id]["error"] = str(e)
        log_data_export_event(
            requester=requester,
            source_type=request.source_type,
            identifiers=request.identifiers,
            outcome=8,
        )


# --- Endpoints ---
@app.post("/extract", response_model=ExtractionResponse)
def extract_data(
    request: ExtractionRequest,
    background_tasks: BackgroundTasks,
    api_key: str = Depends(verify_api_key),
):
    job_id = str(uuid.uuid4())
    jobs[job_id] = {"status": "queued", "files_extracted": []}
    background_tasks.add_task(run_extraction, job_id, request, api_key)
    return ExtractionResponse(job_id=job_id, status="queued")


@app.get("/status/{job_id}", response_model=JobStatus)
def get_status(job_id: str, api_key: str = Depends(verify_api_key)):
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")
    job = jobs[job_id]
    return JobStatus(
        job_id=job_id,
        status=job["status"],
        files_extracted=job.get("files_extracted", []),
    )
Test a DICOM extraction against your local PACS

You need to know a Study Instance UID from your test PACS. You can find one using dcmdump on any of the sample DICOM files you stored earlier.

  • Find a Study Instance UID from one of your sample files by running:
dcmdump ./dicom_store/*.dcm | grep StudyInstanceUID

What does this command do?

The dcmdump tool prints all DICOM tags from a file. Piping through grep StudyInstanceUID extracts just the UID you need for the extraction request.

You should see output like (0020,000d) UI [1.2.840.113619...]. Copy the value inside the square brackets.

  • Make sure dcmqrscp is still running. If you stopped it, restart it by running:
dcmqrscp --config dcmqrscp.cfg
  • In a separate terminal, make sure uvicorn is running:
uvicorn main:app --reload
  • Send a test extraction request using curl. Replace the Study Instance UID with the one you found:
curl -X POST http://localhost:8000/extract \
  -H "Content-Type: application/json" \
  -H "X-API-Key: research-key-001" \
  -d '{"source_type": "dicom", "identifiers": ["[[STUDY_UID=\"your Study Instance UID from dcmdump\"]]}"], "destination": "./staging"}'

What does this request do?

This sends a POST to your /extract endpoint with a valid API key. The body tells the API to use the DICOM adapter, retrieve the study matching your UID, and save the files to a ./staging directory.

You should see a JSON response containing a job_id and "status": "queued".

  • Wait a moment for the background task to run, then check the job status (replace the job ID with yours):
curl -H "X-API-Key: research-key-001" http://localhost:8000/status/[[JOB_ID="your job_id from the previous response"]]

You should see "status": "completed" and a files_extracted array listing .dcm filenames.

  • Verify the files actually landed in the staging directory by running:
ls ./staging

You should see one or more .dcm files named by their SOP Instance UID.

Job status shows "failed"?

Check that dcmqrscp is still running in a separate terminal. If it crashed or was stopped, the association will fail.

Verify your Study Instance UID is correct. A typo or non-existent UID will complete the C-GET with zero results. The files_extracted array will be empty but the job will still show as completed (not failed).

If you see a connection refused error, confirm the port in dicom_adapter.py matches dcmqrscp.cfg (both should be 11112).

help me debug my failed DICOM extraction job

Your DICOM adapter is fully operational. Next up, you will build the SMB adapter so the API can pull waveform and binary files from network shares using the same extraction interface.

Implement the SMB Adapter

Your DICOM adapter handles imaging data from PACS, but hospitals store plenty of non-DICOM data on network file shares. Waveform recordings, binary sensor outputs, and exported reports often live on SMB (Server Message Block) shares accessible across the hospital network.

In this step, you will build a second adapter that copies files from an SMB share into the same staging directory. This means researchers can pull data from both PACS and file shares through the same API endpoint.

In this step, get ready to:
  • Create the SMB adapter that reads files from a network share.
  • Wire the adapter into the extraction endpoint.
  • Test the SMB extraction flow with a POST request.
Create the SMB adapter module

The smbclient high-level API from the smbprotocol package mirrors Python's built-in os module. You can open remote files, scan directories, and check paths using familiar patterns.

  • Create a new file called smb_adapter.py inside the adapters/ folder by copying the following code:
import os
import smbclient


# Configuration - replace with your SMB share details
SMB_SERVER = "your-smb-server"
SMB_USERNAME = "your-username"
SMB_PASSWORD = "your-password"
SMB_SHARE = "your-share-name"


def retrieve_smb(identifiers: list[str], destination: str) -> list[str]:
    """Copy files from SMB share to local staging directory."""
    os.makedirs(destination, exist_ok=True)
    extracted_files = []

    # Register an authenticated session with the SMB server
    smbclient.register_session(
        SMB_SERVER, username=SMB_USERNAME, password=SMB_PASSWORD
    )

    for path in identifiers:
        # Build the full UNC path to the remote resource
        smb_path = rf"\\{SMB_SERVER}\{SMB_SHARE}\{path}"

        if smbclient.path.isdir(smb_path):
            # If the path is a directory, copy all files inside it
            for entry in smbclient.scandir(smb_path):
                if entry.is_file():
                    src = entry.path
                    filename = entry.name
                    dest_path = os.path.join(destination, filename)
                    with smbclient.open_file(src, mode="rb", share_access="rw") as src_file:
                        with open(dest_path, "wb") as dst_file:
                            dst_file.write(src_file.read())
                    extracted_files.append(filename)
        else:
            # Single file - copy it directly
            filename = os.path.basename(path)
            dest_path = os.path.join(destination, filename)
            with smbclient.open_file(smb_path, mode="rb", share_access="rw") as src_file:
                with open(dest_path, "wb") as dst_file:
                    dst_file.write(src_file.read())
            extracted_files.append(filename)

    return extracted_files

What does this code do?

The adapter follows the same interface as your DICOM adapter: it takes a list of identifiers and a destination path, then returns the filenames it extracted.

  • smbclient.register_session() authenticates once with the SMB server. The session is cached and reused for subsequent calls.
  • smbclient.path.isdir() checks whether the remote path is a directory or a single file. This lets researchers request an entire folder or a specific file.
  • share_access="rw" avoids exclusive file locks. Without this, reading a file already open by a clinical system would fail with an access error.

Why share_access="rw" matters in a hospital

Clinical workstations often keep files open while recording data. The default SMB behavior requests exclusive access, which would cause your extraction to fail with NtStatus 0xc0000043. Setting share_access="rw" tells the server you are happy to read while another process writes.

  • Save adapters/smb_adapter.py.
  • Confirm the file exists by running:
ls adapters/

You should see __init__.py, dicom_adapter.py, and smb_adapter.py in the output.

Don't see smb_adapter.py?

  • Make sure you created the file inside the adapters/ folder, not the project root.
  • Check the filename is exactly smb_adapter.py with an underscore, not a hyphen.

Help me find my smb_adapter.py file

Wire the SMB adapter into the extract endpoint

Your main.py currently only knows about the DICOM adapter. You need to import the new SMB adapter and add a branch for source_type: "smb" in the extraction dispatcher.

  • In main.py, find the import section at the top of the file. Locate this line:
from adapters.dicom_adapter import retrieve_dicom
  • Add the SMB adapter import directly below it, so the two import lines look like this:
from adapters.dicom_adapter import retrieve_dicom
from adapters.smb_adapter import retrieve_smb
  • Now scroll down to the run_extraction function. Find the else branch that handles unknown source types:
        if request.source_type == "dicom":
            extracted_files = retrieve_dicom(
                identifiers=request.identifiers,
                destination=request.destination,
            )
        else:
            jobs[job_id]["status"] = "failed"
  • Replace that else: with an elif for SMB followed by a new else, so the full dispatch block looks like this:
        if request.source_type == "dicom":
            extracted_files = retrieve_dicom(
                identifiers=request.identifiers,
                destination=request.destination,
            )
        elif request.source_type == "smb":
            extracted_files = retrieve_smb(
                identifiers=request.identifiers,
                destination=request.destination,
            )
        else:
            jobs[job_id]["status"] = "failed"
            jobs[job_id]["error"] = f"Unknown source_type: {request.source_type}"
            log_data_export_event(
                requester=requester,
                source_type=request.source_type,
                identifiers=request.identifiers,
                outcome=8,
            )
            return

What does this change do?

The dispatcher now routes requests based on source_type. DICOM requests go to the DICOM adapter, SMB requests go to the SMB adapter, and anything else fails with an audit log entry recording the error.

  • Save main.py.

Seeing an IndentationError?

  • The elif must align exactly with the if above it. Both should be indented by 8 spaces (inside the try: block).
  • Make sure you did not accidentally delete the else: block. It should still exist after the elif branch.

Help me fix my indentation in main.py

✔️ Awesome, I've got everything!

Great. Double check you have saved both adapters/smb_adapter.py and main.py.

ⓧ I'd like to double check the full code

Here is the full adapters/smb_adapter.py:

import os
import smbclient


# Configuration - replace with your SMB share details
SMB_SERVER = "your-smb-server"
SMB_USERNAME = "your-username"
SMB_PASSWORD = "your-password"
SMB_SHARE = "your-share-name"


def retrieve_smb(identifiers: list[str], destination: str) -> list[str]:
    """Copy files from SMB share to local staging directory."""
    os.makedirs(destination, exist_ok=True)
    extracted_files = []

    smbclient.register_session(
        SMB_SERVER, username=SMB_USERNAME, password=SMB_PASSWORD
    )

    for path in identifiers:
        smb_path = rf"\\{SMB_SERVER}\{SMB_SHARE}\{path}"

        if smbclient.path.isdir(smb_path):
            for entry in smbclient.scandir(smb_path):
                if entry.is_file():
                    src = entry.path
                    filename = entry.name
                    dest_path = os.path.join(destination, filename)
                    with smbclient.open_file(src, mode="rb", share_access="rw") as src_file:
                        with open(dest_path, "wb") as dst_file:
                            dst_file.write(src_file.read())
                    extracted_files.append(filename)
        else:
            filename = os.path.basename(path)
            dest_path = os.path.join(destination, filename)
            with smbclient.open_file(smb_path, mode="rb", share_access="rw") as src_file:
                with open(dest_path, "wb") as dst_file:
                    dst_file.write(src_file.read())
            extracted_files.append(filename)

    return extracted_files

And here is the full main.py:

import uuid
from fastapi import FastAPI, BackgroundTasks, Depends, HTTPException, Security
from fastapi.security import APIKeyHeader
from pydantic import BaseModel
from adapters.dicom_adapter import retrieve_dicom
from adapters.smb_adapter import retrieve_smb
from audit import log_data_export_event

app = FastAPI(title="Research Data Extraction API", version="1.0.0")

# --- Authentication ---
API_KEY_HEADER = APIKeyHeader(name="X-API-Key")
VALID_API_KEYS = ["research-key-001", "research-key-002"]


def verify_api_key(api_key: str = Security(API_KEY_HEADER)):
    if api_key not in VALID_API_KEYS:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return api_key


# --- Job tracking ---
jobs: dict = {}


# --- Request/Response models ---
class ExtractionRequest(BaseModel):
    source_type: str
    identifiers: list[str]
    destination: str


class ExtractionResponse(BaseModel):
    job_id: str
    status: str


class JobStatus(BaseModel):
    job_id: str
    status: str
    files_extracted: list[str]


# --- Background task ---
def run_extraction(job_id: str, request: ExtractionRequest, requester: str):
    try:
        jobs[job_id]["status"] = "running"
        extracted_files = []

        if request.source_type == "dicom":
            extracted_files = retrieve_dicom(
                identifiers=request.identifiers,
                destination=request.destination,
            )
        elif request.source_type == "smb":
            extracted_files = retrieve_smb(
                identifiers=request.identifiers,
                destination=request.destination,
            )
        else:
            jobs[job_id]["status"] = "failed"
            jobs[job_id]["error"] = f"Unknown source_type: {request.source_type}"
            log_data_export_event(
                requester=requester,
                source_type=request.source_type,
                identifiers=request.identifiers,
                outcome=8,
            )
            return

        jobs[job_id]["status"] = "completed"
        jobs[job_id]["files_extracted"] = extracted_files

        log_data_export_event(
            requester=requester,
            source_type=request.source_type,
            identifiers=request.identifiers,
            outcome=0,
        )
    except Exception as e:
        jobs[job_id]["status"] = "failed"
        jobs[job_id]["error"] = str(e)
        log_data_export_event(
            requester=requester,
            source_type=request.source_type,
            identifiers=request.identifiers,
            outcome=8,
        )


# --- Endpoints ---
@app.post("/extract", response_model=ExtractionResponse)
def extract_data(
    request: ExtractionRequest,
    background_tasks: BackgroundTasks,
    api_key: str = Depends(verify_api_key),
):
    job_id = str(uuid.uuid4())
    jobs[job_id] = {"status": "queued", "files_extracted": []}
    background_tasks.add_task(run_extraction, job_id, request, api_key)
    return ExtractionResponse(job_id=job_id, status="queued")


@app.get("/status/{job_id}", response_model=JobStatus)
def get_status(job_id: str, api_key: str = Depends(verify_api_key)):
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")
    job = jobs[job_id]
    return JobStatus(
        job_id=job_id,
        status=job["status"],
        files_extracted=job.get("files_extracted", []),
    )
Test the SMB adapter with a POST request

To confirm the wiring works, you will send a test extraction request with source_type: "smb". If you have a real SMB share available, update the configuration constants in smb_adapter.py first.

  • Make sure uvicorn is running. If you stopped it earlier, start it again by running:
uvicorn main:app --reload
  • Open a second terminal window and send a test request by running:
curl -X POST http://localhost:8000/extract \
  -H "Content-Type: application/json" \
  -H "X-API-Key: research-key-001" \
  -d '{"source_type": "smb", "identifiers": ["waveforms/patient_001.bin"], "destination": "./staging/smb_test"}'

You should see a JSON response with a job_id and status: "queued".

What if I don't have a real SMB share?

If you do not have an SMB share to connect to, the job will move to failed status because the connection cannot be established. That is expected.

The important thing to verify is that the API accepts the request, returns a job ID, and dispatches to the SMB adapter (not the unknown-source error path). Check the job status to confirm the error message mentions the SMB connection, not an unknown source type.

  • Check the job status by running (replace the job ID with the one from your response):
curl -H "X-API-Key: research-key-001" http://localhost:8000/status/[[JOB_ID="your-job-id-from-the-previous-response"]]

If you have a real SMB share configured, the status should show completed with filenames in the files_extracted array. If no share is available, you will see failed with an SMB connection error.

Getting "Unknown source_type" in the error?

  • This means the elif branch was not added correctly. Open main.py and verify you have elif request.source_type == "smb": between the DICOM if and the else.
  • Make sure uvicorn reloaded after your last save. Check the terminal for a "Reloading..." message.

Help me debug my SMB adapter wiring

Your API now has two fully wired adapters. Next up, you will run the full end-to-end integration test to confirm both sources work together with authentication and audit logging.

Run End-to-End Integration Tests

You now have a complete multi-source extraction API with both adapters wired in and audit logging active. The real test of any integration service is whether data actually flows end-to-end from source to staging.

In this step, you will submit extraction requests to your running API, verify that files land in the staging directory, confirm that authentication rejects unauthorized callers, and inspect the ATNA audit log for compliance records.

In this step, get ready to:
  • Submit DICOM and SMB extraction requests using curl.
  • Verify extracted files and job completion status.
  • Confirm API key authentication rejects invalid requests and ATNA audit messages are logged.
Submit a DICOM extraction request

Your FastAPI server should still be running from the previous step. If not, start it again before continuing.

  • If your uvicorn server is not running, restart it by running:
uvicorn main:app --reload
  • Open a separate terminal window for running test commands.
  • Make sure your dcmqrscp process is also still running in its own terminal.

The POST /extract endpoint expects a JSON body with a source_type, a list of identifiers, and a destination path. For DICOM requests, identifiers are Study Instance UIDs.

  • Submit a DICOM extraction request by running this curl command in your test terminal:
curl -X POST http://localhost:8000/extract \
  -H "Content-Type: application/json" \
  -H "X-API-Key: research-key-001" \
  -d '{"source_type": "dicom", "identifiers": ["[[STUDY_UID=\"your Study Instance UID from the sample DICOM file\"]]" ], "destination": "./staging/dicom"}'

Where do I find my Study Instance UID?

The Study Instance UID comes from the sample DICOM file you uploaded to dcmqrscp in Step 2. You can inspect any .dcm file to find it by running:

dcmdump ./dicom_store/*.dcm | grep StudyInstanceUID

Copy the UID value (e.g. 1.2.840.113619.2.55.3.123456) and use it in your curl command.

  • You should see a JSON response with a job_id and status queued.

What should I see?

The response looks like this:

{"job_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890", "status": "queued"}

Your job_id will be a different UUID. Copy it for the next command.

Getting a connection refused error?

Make sure uvicorn is running in another terminal. Check that you are using port 8000 (the default FastAPI port) in the curl URL, not port 11112 (that is the PACS port).

Help me debug my curl connection error

Verify files and job status

The extraction runs as a background task, so it completes shortly after the initial response. You can check on the job using the status endpoint.

  • Check the job status by running this command (replace the job ID with the one from your response):
curl -X GET http://localhost:8000/status/[[JOB_ID="your job_id from the previous response"]] \
  -H "X-API-Key: research-key-001"

What should I see?

The response shows "status": "completed" and a files_extracted array listing the .dcm filenames that were retrieved from the PACS.

  • Confirm the DICOM files landed in the staging directory by running:
ls ./staging/dicom
  • You should see one or more .dcm files named by their SOP Instance UID.

Status shows "failed" instead of "completed"?

Check that dcmqrscp is still running in its own terminal. The DICOM adapter connects to port 11112 and needs the PACS to be reachable.

Also verify that the Study Instance UID you used actually exists in your dcmqrscp store. Run dcmdump ./dicom_store/*.dcm | grep StudyInstanceUID to confirm.

Help me debug my failed DICOM extraction job

If you have an SMB share configured in adapters/smb_adapter.py, you can also test the SMB adapter.

  • Submit an SMB extraction request by running:
curl -X POST http://localhost:8000/extract \
  -H "Content-Type: application/json" \
  -H "X-API-Key: research-key-001" \
  -d '{"source_type": "smb", "identifiers": ["[[SMB_PATH=\"path/to/file-on-share\"]]" ], "destination": "./staging/smb"}'

No SMB share available?

If you do not have an SMB share configured, the request will still return a job_id with status queued. The job status will show failed because the connection cannot be established. This is expected. The important thing is that the API dispatched correctly and an ATNA audit message was still logged for the failed attempt.

Verify authentication and ATNA audit logging

A secure research API must reject unauthorized requests. Your authentication dependency checks the X-API-Key header against the configured keys.

  • Test a request with no API key by running:
curl -X POST http://localhost:8000/extract \
  -H "Content-Type: application/json" \
  -d '{"source_type": "dicom", "identifiers": ["1.2.3.4"], "destination": "./staging/test"}'
  • You should see a 403 response with "detail": "Not authenticated".
  • Test a request with an invalid API key by running:
curl -X POST http://localhost:8000/extract \
  -H "Content-Type: application/json" \
  -H "X-API-Key: wrong-key-999" \
  -d '{"source_type": "dicom", "identifiers": ["1.2.3.4"], "destination": "./staging/test"}'
  • You should see a 401 response with "detail": "Invalid API key".

Why 403 without a key but 401 with an invalid key?

FastAPI's APIKeyHeader returns a 403 when the header is missing entirely (no credentials provided). Your custom verify_api_key function raises a 401 when the key is present but does not match. Both outcomes confirm the authentication layer is working.

Every extraction event (successful or failed) should generate an IHE ATNA audit message in your log file.

  • Inspect the ATNA audit log by running:
cat atna_audit.log
  • You should see XML audit messages containing <AuditMessage> elements with EventActionCode="R" for each extraction you ran.

What does the audit log contain?

Each XML message follows the DICOM PS3.15 A.5 format. It records who requested the data (ActiveParticipant with the API key as UserID), what was accessed (ParticipantObjectIdentification with your Study Instance UID), and the outcome (EventOutcomeIndicator 0 for success, 8 for failure).

Successful DICOM extractions show outcome 0. Failed SMB attempts (if you do not have a share configured) show outcome 8. Both are recorded for compliance.

Audit log file is empty or missing?

The atna_audit.log file is created in the directory where you started uvicorn. Make sure you are running cat from the research-extraction-api folder.

If the file exists but is empty, check that your extraction requests included a valid API key. Only requests that pass authentication reach the extraction logic where audit logging happens.

Help me find my ATNA audit log

Your multi-source extraction API is fully operational. It authenticates callers, dispatches to the correct source adapter, saves files to staging, tracks job progress, and logs every access event in IHE ATNA format. Next up is the Secret Mission, where you can add LDAP authentication for production deployment.

Secret mission

Add LDAP/Active Directory Authentication

API keys work great for development, but hospital production environments require user-level authentication. Upgrade your extraction API to validate credentials against Active Directory using LDAP, making it deployment-ready for enterprise networks.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing costs.

Resources you used:

  • Python virtual environment (venv folder inside your project directory).
  • FastAPI application files (main.py, audit.py, adapters/).
  • dcmqrscp local test PACS process and configuration file (dcmqrscp.cfg).
  • DICOM storage directory (dicom_store/) with sample DICOM files.
  • ATNA audit log file (atna_audit.log).
  • Staging directory with extracted files.

Keep everything running

No action needed. Choose this if you are still actively building or plan to extend the project with the Secret Mission.

  • Leave your virtual environment activated and your project files in place.
  • Keep dcmqrscp running if you want to continue testing DICOM retrieval.
  • Your uvicorn dev server will stop when you close the terminal, but you can restart it anytime.

Pause - I'll come back to this later

Shut down running processes to free up system resources. Your files stay on disk for when you return.

  • Stop the uvicorn dev server by pressing Ctrl+C in its terminal window.
  • Stop the dcmqrscp process by pressing Ctrl+C in its terminal window.
  • Deactivate your virtual environment by running deactivate in your terminal.
  • When you come back, reactivate the virtual environment, restart dcmqrscp, and start uvicorn again.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes everything you built in this project.

  • Stop the uvicorn dev server and dcmqrscp process by pressing Ctrl+C in each terminal window.
  • Delete the entire project folder from your Desktop. This removes all Python files, the virtual environment, the DICOM storage directory, audit logs, and staging files.
  • Python and DCMTK remain installed on your system. Uninstall them separately through your system's package manager if you no longer need them.

Nice Work!

Nice Work!

You just built a production-grade, multi-source research data extraction API with full IHE ATNA audit compliance. Researchers can now pull imaging and waveform data from heterogeneous hospital systems through a single secure endpoint.

You've learned how to:

  • Build a FastAPI service with background task processing, job status tracking, and API key authentication to secure extraction endpoints.
  • Create DICOM and SMB adapters using the adapter pattern to extract data from a PACS via C-GET and from network shares into a unified staging directory.
  • Implement an IHE ATNA audit logger that produces DICOM PS3.15 A.5 compliant XML audit messages and transmits them via RFC 5424 syslog.
  • Added LDAP/Active Directory authentication for production deployment with environment-switchable auth modes.

Ready to quiz yourself?