Building a Data-Analysis Assistant

Ma Mahalakshmi V Updated 16 Sep 2026
7 min read ·Lesson 92 of 224

From Individual Calls to a Reusable Component

Every previous lesson in this unit demonstrated one request at a time: upload a file, ask a question, extract a result. A real application needs these pieces wired together behind a stable interface — something a web endpoint, a CLI tool, or a chat UI can call repeatedly across a user's session without re-deriving the upload, container-reuse, and extraction logic on every call site. This lesson builds that component: a small DataAnalysisAssistant class that owns a session's uploaded files, its container, and the conversation chain, and exposes a simple ask() method.

Designing the Interface First

Before writing the implementation, it helps to decide what the class should look like from the outside, since that shapes every decision inside it:

assistant = DataAnalysisAssistant()
assistant.upload_dataset("quarterly_sales.csv")
answer = assistant.ask("What were total sales by region?")
answer2 = assistant.ask("Now break that down by month for the top region.")
charts = assistant.get_generated_files()

This interface hides three things a caller should not have to think about on every use: which files are currently attached, which container and response ID the conversation is currently chained to, and how to extract generated files from the raw response. Each ask() call should feel like a single, stateless-looking function call from the outside, while internally maintaining the state needed for the follow-up question to correctly reuse the loaded data (the pattern from Lesson 2 and Lesson 4).

Implementation

from openai import OpenAI


class DataAnalysisAssistant:
    """Wraps code interpreter to provide a simple, stateful interface for
    multi-turn data analysis over one or more uploaded datasets."""

    def __init__(self, client: OpenAI | None = None, model: str = "gpt-5.6-terra"):
        self.client = client or OpenAI()
        self.model = model
        self._file_ids: list[str] = []
        self._previous_response_id: str | None = None
        self._last_response = None

    def upload_dataset(self, path: str) -> str:
        uploaded = self.client.files.create(file=open(path, "rb"), purpose="assistants")
        self._file_ids.append(uploaded.id)
        return uploaded.id

    def ask(self, question: str) -> str:
        tool_config = {"type": "code_interpreter", "container": {"type": "auto"}}
        if self._file_ids and self._previous_response_id is None:
            tool_config["container"]["file_ids"] = self._file_ids

        kwargs = {
            "model": self.model,
            "tools": [tool_config],
            "input": question,
        }
        if self._previous_response_id:
            kwargs["previous_response_id"] = self._previous_response_id

        response = self.client.responses.create(**kwargs)
        self._previous_response_id = response.id
        self._last_response = response
        return response.output_text

    def get_generated_files(self) -> list[dict]:
        if self._last_response is None:
            return []
        return extract_generated_files(self._last_response, self.client)

    def cleanup(self) -> None:
        for file_id in self._file_ids:
            self.client.files.delete(file_id)
        self._file_ids = []

Walking Through the Design Decisions

File IDs are only attached on the first call. The condition if self._file_ids and self._previous_response_id is None deliberately attaches file_ids to the container only when there is no prior response to chain from — that is, on the very first question. Every subsequent ask() call omits file_ids because the files are already loaded into the container from the first call, and previous_response_id carries that container's state forward, exactly as demonstrated in Lesson 2. Attaching the same file IDs again on every call would be redundant at best and, depending on platform behavior, could unnecessarily trigger a fresh load of the file rather than reusing the already-parsed dataframe.

previous_response_id is only added to kwargs when it exists. This lets ask() work correctly on both the very first call (no prior response yet) and every call after, using the same method body — a common and useful pattern for wrapping a conversational API where the first turn and later turns need slightly different arguments but should share one code path.

get_generated_files() reuses the extract_generated_files helper from Lesson 6 rather than reimplementing extraction logic inside the class. This is deliberate: extraction is a pure function of a response object, and keeping it as a standalone function (rather than a method tightly coupled to this class) means it can be tested and reused independently — which is exactly what the test below does.

cleanup() is a separate, explicit method rather than automatic. Deleting uploaded files the moment an ask() call finishes would break the entire point of the class, since follow-up questions need those files to still exist for container recreation scenarios. Cleanup belongs at the end of a session's lifetime, called explicitly by whatever code owns the assistant's lifecycle — a request handler's finally block, a context manager, or an explicit "end session" action in a UI.

Using It as a Context Manager

For code that wants a guarantee that cleanup always happens, wrapping the class with context-manager support is a natural extension:

class ManagedDataAnalysisAssistant(DataAnalysisAssistant):
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.cleanup()
        return False


with ManagedDataAnalysisAssistant() as assistant:
    assistant.upload_dataset("quarterly_sales.csv")
    print(assistant.ask("What is the total revenue across all regions?"))

__exit__ returning False means any exception raised inside the with block is not suppressed — cleanup happens, but the caller still finds out something went wrong, rather than the error being silently swallowed. This is the correct default for cleanup logic: clean up resources unconditionally, but never hide a real failure from the caller by accident.

Testing the Assistant Without Live API Calls

Following this course's dependency-injection testing pattern, the assistant is tested by injecting a fake client rather than calling the real API:

class FakeFilesResource:
    def __init__(self):
        self.created = []
        self.deleted = []

    def create(self, file, purpose):
        file_id = f"file_{len(self.created)}"
        self.created.append(file_id)
        class FakeFile:
            id = file_id
        return FakeFile()

    def delete(self, file_id):
        self.deleted.append(file_id)


class FakeResponsesResource:
    def __init__(self):
        self.calls = []

    def create(self, **kwargs):
        self.calls.append(kwargs)
        class FakeResponse:
            id = f"resp_{len(self.calls)}"
            output_text = f"answer to: {kwargs['input']}"
            output = []
        return FakeResponse()


class FakeClient:
    def __init__(self):
        self.files = FakeFilesResource()
        self.responses = FakeResponsesResource()


def test_assistant_attaches_files_only_on_first_call():
    fake_client = FakeClient()
    assistant = DataAnalysisAssistant(client=fake_client)
    assistant.upload_dataset("dummy_path.csv")

    first_answer = assistant.ask("first question")
    second_answer = assistant.ask("second question")

    assert first_answer == "answer to: first question"
    assert second_answer == "answer to: second question"

    first_call, second_call = fake_client.responses.calls
    assert "file_ids" in first_call["tools"][0]["container"]
    assert "file_ids" not in second_call["tools"][0]["container"]
    assert "previous_response_id" not in first_call
    assert second_call["previous_response_id"] == "resp_1"

    print("PASS: file_ids attached only on first call, chaining used afterward")


def test_cleanup_deletes_all_uploaded_files():
    fake_client = FakeClient()
    assistant = DataAnalysisAssistant(client=fake_client)
    assistant.upload_dataset("a.csv")
    assistant.upload_dataset("b.csv")

    assistant.cleanup()

    assert fake_client.files.deleted == ["file_0", "file_1"]
    assert assistant._file_ids == []
    print("PASS: cleanup deletes every uploaded file and clears local state")


test_assistant_attaches_files_only_on_first_call()
test_cleanup_deletes_all_uploaded_files()

FakeFilesResource and FakeResponsesResource record every call they receive instead of hitting a real API, which lets the tests assert on exactly how DataAnalysisAssistant uses its client — specifically, that file_ids appears only on the first responses.create call and that previous_response_id correctly chains the second call to the first response's ID. This verifies the class's internal logic precisely, without needing network access, an API key, or any nondeterminism from an actual model response — the fake output_text is deterministic and directly reflects the input it was given, making the assertions exact rather than approximate.

Note that upload_dataset("dummy_path.csv") works against the fake client without the file needing to actually exist on disk, because FakeFilesResource.create never calls open() itself — a small but important detail: the real DataAnalysisAssistant.upload_dataset does call open(path, "rb") before handing it to self.client.files.create, so a fully faithful unit test would need either a real temporary file or a slightly different injection point. In practice, most teams solve this by injecting an already-open file-like object or by using a temporary file created within the test itself, keeping the fake client focused purely on faking the network boundary.

Common Mistakes

Building a new DataAnalysisAssistant instance per question instead of per session. This defeats the entire purpose of the class — a fresh instance has no previous_response_id and no memory of uploaded files, so every "follow-up" question would actually re-upload files and start a brand-new, unrelated conversation.

Forgetting to call cleanup() at the end of a session, leaking uploaded files exactly as described in Lesson 3. Wrapping session-ending code paths (including error paths) in a finally block or a context manager, as shown above, prevents this.

Testing this class only through the real API during development. This makes tests slow, costly, and flaky (subject to model nondeterminism and network conditions). The fake-client pattern above should be the default way this logic is verified during day-to-day development, with real API calls reserved for periodic integration testing.

Best Practices

Keep the assistant class thin and delegate reusable logic (like file extraction) to standalone functions, so those functions can be tested and reused independently of the class that happens to call them.

Expose an explicit cleanup() method and pair it with context-manager support so that both explicit and automatic resource management styles are available to callers, matching how they structure the rest of their application.

Design the fake objects used in tests to record what they were called with, not just to return canned data. Asserting on the actual arguments passed to responses.create (as the tests above do) verifies the class's real behavior — correct container and chaining logic — rather than merely confirming it doesn't crash.

0 Comments

Reviewed before they appear

No comments yet.

OpenAI SDK
Introduction to the OpenAI SDK Setting Up Python Creating an API Key Your First Call — client.responses.create() and response.output_text Understanding Billing, Credits, and What a Request Costs Why Responses Replaced Chat Completions Anatomy of a Request: model, input, and instructions Anatomy of a Response: The Typed output Array, Not Just Text Roles: User, Assistant, and Developer/System Choosing a Model, and Reading the Models Page Instead of Memorizing Names Instructions vs. Input Writing Prompts That Get Consistent Results Few-Shot Examples Reasoning Models and the reasoning Parameter Debugging a Prompt That Misbehaves Why Streaming Matters for User Experience stream=True and Iterating Over Events Handling the Event Types You Actually Care About Background Mode for Long-Running Jobs Project — Add Live Streaming to Your Chatbot The Problem With Parsing Free Text JSON Schema and Strict Mode Pydantic Models With the SDK's Parse Helpers Handling Refusals and Validation Failures Project — A Resume-to-JSON Extractor Working With input_image input_file, PDFs, and the Files API Image Generation Speech-to-Text and Text-to-Speech Project: A PDF Question-Answering Script What Function Calling Is Defining a Tool Schema The Full Loop Multiple Tools Errors, Timeouts, and Untrusted Arguments Project: A Weather Assistant Web Search File Search and Vector Stores Code Interpreter Remote MCP Servers and Connectors Project: A Research Assistant What an Embedding Is, Without the Maths Generating and Storing Embeddings Similarity Search From Scratch Hosted Vector Stores vs. Rolling Your Own A Small RAG App Over a Folder of Notes Agents vs. a Single API Call — When You Need One pip install openai Giving Agents Tools Handoffs and Multi-Agent Triage Guardrails and Approvals Tracing and Observing What Your Agent Did A Multi-Agent Support Desk Error Codes and What Each One Means Retries, Timeouts, and Backoff Rate Limits and Spend Limits Prompt Caching and Cost Optimisation The Batch API for Bulk Work Async Clients and Concurrency Moderation and Safety Best Practices Designing the App Backend With FastAPI Streaming to a Simple Frontend Deploying and a Cost/Safety Checklist Why Web Search Is Useful for Current Information Using the Web Search Tool with the Responses API Configuring Search Behavior for Application Use Cases Understanding Citations and Source Attribution Where to Go Next Building a Research Assistant with Web Search Combining Web Search with Structured Outputs Handling Conflicting or Low-Quality Web Sources Reducing Unsupported Claims with Grounded Generation Testing Freshness-Sensitive AI Answers Production Considerations for Web-Grounded Applications Understanding File Search and Retrieval-Augmented Generation Creating and Organizing Vector Stores Uploading Documents for Retrieval Connecting Vector Stores to Responses API Requests Designing Document Metadata and Filtering Strategies Building a PDF Question-Answering Application Improving Retrieval Quality With Better Document Preparation Handling Missing Evidence and Retrieval Failures Combining File Search With Web Search Building a Production Knowledge-Base Assistant What the Code Interpreter Tool Is Designed For Running Python-Based Analysis Through the OpenAI SDK Uploading Datasets for Analysis Analyzing CSV and Spreadsheet Data Generating Charts and Data Summaries Handling Generated Files and Downloadable Artifacts Building a Data-Analysis Assistant Combining Code Execution with Structured Outputs Validating Generated Calculations and Results Security and Sandbox Considerations for Code Execution Understanding Multimodal Input with the OpenAI SDK Sending Images to a Model Image Analysis from URLs and Uploaded Files Extracting Text and Information from Screenshots Building an Image-Question-Answering Application Combining Image Input with Structured Output Analyzing Multiple Images in One Request Handling Image Quality and Input Limitations Designing Multimodal Prompts for Reliable Results Building a Practical Vision-Powered Python Application Understanding Speech-to-Text and Text-to-Speech Workflows Transcribing Audio with the OpenAI SDK Working with Uploaded Audio Files Handling Timestamps and Transcription Metadata Building a Meeting Transcription Workflow Generating Spoken Responses from Text Handling Long Audio and Processing Failures Combining Audio with Text and Tool Calling Building an End-to-End Python Voice Application What Embeddings Are and When to Use Them Generating Embeddings With the OpenAI API Preparing Text for Embedding Comparing Vectors With Cosine Similarity Building a Simple Semantic Search Engine in Python Storing Embeddings in a Database Metadata Filtering for Semantic Search Chunking Strategies for Better Retrieval Evaluating Semantic Search Quality Building a Document Similarity Application When Batch Processing Makes Sense Designing Large-Volume AI Processing Pipelines Using Asynchronous Python with the OpenAI SDK Running Concurrent Requests Safely Controlling Concurrency and Avoiding Rate Limits Tracking Batch Job Progress Handling Partial Failures in Bulk Workloads Retrying Failed Items Without Duplicating Successful Work Designing Resumable AI Processing Jobs Building a Production Batch-Processing Pipeline Batch Processing Makes Sense Large-Scale AI Processing Pipelines Async Python with OpenAI SDK Safe Concurrent Requests Concurrency & Rate Limits Batch Progress Tracking Partial Failure Handling Safe Retry Handling Resumable AI Jobs Production Batch Pipeline System–User Data Separation Reusable App Instructions Prompt Templates & Variables Extraction & Classification Prompts Summarization & Transformation Prompts Explicit Output Requirements Prompt Version Management Prompt Testing & Evaluation Reusable Python Prompt Library API Key Security Secure API Key Storage Secure Secret Management Prompt Injection Prevention Trusted vs. Untrusted Content Tool Argument Validation Sensitive Data Handling Secure Logging AI Action Authorization Production AI Security Checklist Why AI Applications Need Evaluation Beyond Unit Tests Unit Testing OpenAI SDK Integration Code Mocking API Responses in Python Tests Testing Structured Outputs Against Schemas Testing Tool-Calling Workflows Building a Small Evaluation Dataset Measuring Accuracy, Consistency, and Failure Rates Regression Testing Prompts and Model Changes Human Evaluation Versus Automated Evaluation Creating a Repeatable Evaluation Pipeline AI Request Monitoring Token Cost Management Usage Metrics Design Reducing Model Calls Prompt & Context Optimization Model Selection & Optimization AI Caching Strategies Interactive Latency Optimization Usage Dashboards & Budget Alerts Performance & Cost Checklist Every API Call Starts Fresh Fixing API Statelessness Server-Side Conversation Memory Limits of Response Chaining What We're Building Conversation Memory Challenges Preparing an OpenAI SDK Application for Deployment Environment-Specific Configuration for Development and Production Deploying a Python AI Service with Docker Container Health Checks and Startup Configuration Managing Secrets in Cloud Deployments Background Workers for Long-Running AI Tasks Queues and Asynchronous Job Architectures Scaling AI Workloads Horizontally Monitoring Production Incidents and Failures Production Deployment Checklist for OpenAI SDK Applications Reusable OpenAI Service Classes AI Client Dependency Injection Typed AI Responses Python Configuration Management AI Request Decorators Centralized AI Error Handling Clean SDK Abstractions Reusable OpenAI Utilities Internal AI Python Libraries SDK Integration Maintenance Production AI Chatbot Document Q&A System Web Research Assistant Customer Support Agent AI Data Analysis Assistant Image Analysis App Meeting Transcription & Summary Semantic Document Search Multi-Tool AI Agent Production OpenAI SDK App Why "It Looked Fine When I Tested It" Isn't Enough Timing Note Status Note Pre-Decision Status Note Current Availability Note
Ask about this post
AI Ask about this post

Ask questions about Building a Data-Analysis Assistant and get answers drawn from it.

Signed-in readers only.