Ingest PRDs with Azure AI

Introduction

30 Second Summary

A product requirements document can reach a team through a person or another service. When each route handles the attachment differently, the same file can produce inconsistent results.

In this project, you will build a local FastAPI prototype that accepts PRD attachments through a manual upload or an API request for an existing source blob. The shared processing pipeline uses Azure AI Document Intelligence Layout to create Markdown before saving consistent artifacts in Azure Blob Storage.

What You'll Build

You can demo two PRD attachment routes converging on one processor that produces matching Markdown and manifest results.

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

• A manual upload path that turns a PRD submitted through interactive API documentation into a processed manifest.
• An API-triggered path that retrieves an existing source blob before producing the same result shape.
• A consistent artifact set where you can inspect the original document copy, generated Markdown, deferred extraction JSON, and manifest in Blob Storage.
• Secret Mission: add failure manifests so a downstream processing exception leaves an observable record in Blob Storage.

Are there any prerequisites?

You need a Windows computer with access to BeSure's existing Azure AI Document Intelligence Layout resource and Blob Storage flow. Keep the existing endpoint, key, Storage connection string, source container, and output container available.

The project covers local Python and Visual Studio Code setup.

Before We Start

This first checkpoint locks in the prototype's architecture before any hands-on work begins. Both manual multipart uploads and API-triggered Azure Blob Storage retrieval will normalize into one AttachmentEnvelope before calling one shared AttachmentProcessor during integration with BeSure's existing PRD attachment flow.

Set Up the Windows Prototype

Your shared ingestion design needs a safe place to run before it connects to BeSure's deployed application. A local FastAPI prototype gives you an isolated testing boundary.

A reproducible Windows environment keeps every test separate from the deployed service. Python runs the prototype. Visual Studio Code keeps its files together.

In this step, get ready to:

• Verify the required Windows tools.
• Build the pinned local environment.
• Run the prototype health endpoint.

Prepare the Windows tools

The prototype uses Python 3.13.15 so its runtime stays consistent. Start by checking the version available on your computer.

• Press the Windows key to open Windows search.
• Type PowerShell into the search bar.
• Press Enter to open Windows PowerShell.
• Check the installed Python version by running this command:

python --version

What does this check do?

This command asks the Python executable on your Windows path to report its version. The result determines which setup path you need.

✔️ I see the required version

Python 3.13.15 is ready for the prototype. Keep this PowerShell window available until Visual Studio Code is open.

ⓧ I see an older version

Your current Python installation is older than the pinned release. Install Python 3.13.15 before creating the virtual environment.

• Open the official Python 3.13.15 release page.
• Download the recommended 64-bit Windows installer.
• Start the installer.
• Enable the installer option that adds Python to your Windows path.
• Complete the installation.
• Close the current PowerShell window.
• Open a new Windows PowerShell window through Windows search.
• Run the version check again.

ⓧ Command not found

Windows cannot currently find Python. Install the pinned release before continuing.

• Open the official Python 3.13.15 release page.
• Download the recommended 64-bit Windows installer.
• Start the installer.
• Enable the installer option that adds Python to your Windows path.
• Complete the installation.
• Open a new Windows PowerShell window through Windows search.
• Run the version check again.

Still missing the required Python version?

Close every open PowerShell window after the installation. A new window loads the updated Windows path.

If the older version remains active, rerun the official installer with its Windows path option enabled.

Help me fix the Python version detected by Windows.

Visual Studio Code gives you an editor with an integrated terminal. Check whether Windows can find it before creating the prototype workspace.

• Press the Windows key to open Windows search.
• Type Visual Studio Code into the search bar.
• Press Enter.

✔️ Visual Studio Code opens

Visual Studio Code is ready. Keep the window open for the workspace setup.

ⓧ Visual Studio Code is missing

Install Visual Studio Code with the recommended Windows User Setup before continuing.

• Open the official Visual Studio Code Windows setup page.
• Download the Windows User Setup installer.
• Start the installer.
• Approve the Windows installation prompt if it appears.
• Complete the installation.
• Open Visual Studio Code through Windows search.

Visual Studio Code not opening?

Confirm that the installer completed under your Windows account. Restart Windows search after the installation finishes.

Help me open Visual Studio Code on Windows.

The workspace belongs on your Desktop so you can find every local artifact during testing. Its app subfolder becomes the importable service package.

• Click File in the Visual Studio Code menu bar.
• Select Open Folder.
• Select your Desktop in the folder picker.
• Click New Folder.
• Enter prd-ingestion-prototype as the folder name.
• Select the prd-ingestion-prototype folder.
• Click Select Folder.

You will see prd-ingestion-prototype at the top of the Visual Studio Code Explorer sidebar.

• Click the new-folder control at the top of the Explorer sidebar.
• Enter app as the folder name.

You will see the empty app folder inside prd-ingestion-prototype.

Why use an app package?

The app folder separates importable service code from local configuration. This structure lets Uvicorn resolve app.main from the workspace folder.

Build the pinned environment

A virtual environment keeps the prototype's packages inside .venv. The pinned requirements.txt file makes that environment reproducible.

• Create requirements.txt inside prd-ingestion-prototype with the new-file control in the Explorer sidebar.
• Fill requirements.txt by pasting this package list:

azure-ai-documentintelligence==1.0.2
azure-core==1.41.0
azure-storage-blob==12.30.2
fastapi==0.141.1
pydantic==2.13.5
python-multipart==0.0.32
uvicorn==0.53.0

What does this file control?

• The Azure packages provide document analysis plus Blob Storage access.
• FastAPI supplies the HTTP application framework.
• Pydantic validates request data.
• The multipart package enables file uploads.
• Uvicorn runs the local application.

• Save requirements.txt.
• Confirm that requirements.txt remains listed beside the app folder in the Explorer sidebar.

Requirements file showing errors?

Check every package pin against the reference block. A missing equals sign changes how the package installer reads the requirement.

Help me compare my requirements file with the project pins.

• Create .gitignore inside prd-ingestion-prototype with the new-file control in the Explorer sidebar.
• Fill .gitignore by pasting these local exclusions:

.venv/
__pycache__/
*.pyc
settings.local.json

What stays out of source control?

This file excludes the virtual environment plus generated Python cache files. It also excludes the local settings file that holds your existing Azure credentials.

• Save .gitignore.
• Confirm that .gitignore appears in the Explorer sidebar.

Cannot see the ignore file?

Confirm that the filename begins with a full stop. Remove any extra text extension that Windows added.

Help me correct the local ignore filename.

An empty app/__init__.py marks app as a Python package. The file must remain empty.

• Select the app folder in the Explorer sidebar.
• Create __init__.py with the new-file control.
• Leave app/__init__.py empty.

You will see app/__init__.py inside the app folder. Opening it shows a blank editor.

• Click Terminal in the Visual Studio Code menu bar.
• Select New Terminal.
• Create .venv by running this command:

python -m venv .venv

What does this command create?

Python creates an isolated interpreter plus package directory inside .venv. The prototype can use that interpreter without changing your system packages.

• Confirm that .venv appears in the Explorer sidebar.

Virtual environment missing?

Check that the terminal prompt points to the prd-ingestion-prototype folder. Confirm that the earlier version check reports Python 3.13.15.

Help me create the Windows virtual environment.

The explicit virtual environment interpreter sends every dependency into this project. Activating the environment is optional when you use its full interpreter path.

• Install the pinned dependencies by running this command:

.\.venv\Scripts\python.exe -m pip install -r requirements.txt

Why use the virtual environment interpreter?

The .\.venv\Scripts\python.exe path selects the interpreter inside this workspace. Every pinned package is installed into .venv.

• Wait for the installation command to return to the PowerShell prompt.
• Confirm that the output finishes without a package installation error.

The dependency install is complete. Your isolated Python environment can now run the pinned FastAPI application.

Dependency installation failed?

Confirm that requirements.txt is saved inside prd-ingestion-prototype. Check that the command starts with the interpreter inside .venv.

Help me diagnose the failed dependency installation.

Configure and run the health check

The prototype needs a shareable settings template plus a private local copy. The private copy holds credentials for your existing Azure AI Document Intelligence resource and Azure Blob Storage account.

• Create settings.local.json.example inside prd-ingestion-prototype with the new-file control in the Explorer sidebar.
• Fill settings.local.json.example by pasting this template:

{
  "document_intelligence_endpoint": "your-document-intelligence-endpoint",
  "document_intelligence_key": "your-document-intelligence-key",
  "storage_connection_string": "your-storage-connection-string",
  "output_container": "your-existing-output-container"
}

What does this template describe?

• The endpoint plus key identify your existing Document Intelligence resource.
• The connection string gives the prototype access to the existing Storage account.
• The output container identifies where later processing artifacts belong.

• Save settings.local.json.example.
• Confirm that the template appears in the Explorer sidebar.

Template showing a JSON error?

Check the commas after the first three entries. Confirm that every key plus placeholder remains inside double quotes.

Help me fix the settings template syntax.

This next file holds live Azure credentials. Keep it on your computer so those values never enter source control.

• Right-click settings.local.json.example in the Explorer sidebar.
• Select Copy.
• Right-click an empty area in the Explorer sidebar.
• Select Paste.
• Right-click the copied settings file.
• Select Rename.
• Enter settings.local.json as the new filename.
• Open settings.local.json from the Explorer sidebar.

You will see the four placeholder values copied from settings.local.json.example.

• Replace your-document-intelligence-endpoint with your existing Document Intelligence endpoint.
• Replace your-document-intelligence-key with your existing Document Intelligence key.
• Replace your-storage-connection-string with your existing Storage connection string.
• Replace your-existing-output-container with your existing output container name.
• Save settings.local.json.

Keep local settings private

Keep settings.local.json on your computer. Never paste its key or connection string into a screenshot or shared message.

The matching .gitignore entry excludes this local secret file from source control.

A health route proves the local web service can start before document ingestion is added. It gives you one small response that tests the environment.

• Select the app folder in the Explorer sidebar.
• Create main.py with the new-file control.
• Build the health-only application by pasting this code into app/main.py:

from fastapi import FastAPI

app = FastAPI(title="BeSure PRD Attachment Ingestion Prototype")


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}

What does this application do?

• The FastAPI instance defines the local application that Uvicorn imports.
• The /health route accepts a browser request.
• The health() function returns a small JSON status response.

• Save app/main.py.
• Confirm that main.py appears below __init__.py in the app folder.

Python file showing a syntax error?

Check the quotation marks around the application title plus route. Confirm that the return value keeps its surrounding braces.

Help me compare my health application with the reference.

✔️ Awesome, I've got everything!

Your project files are ready. Save every open file before starting the local server.

ⓧ I'd like to double check the full code

Compare each local file with its reference below. Keep your real credentials inside settings.local.json.

.gitignore:

.venv/
__pycache__/
*.pyc
settings.local.json

What should match?

These four entries exactly exclude the virtual environment plus Python cache files plus local settings.

app/__init__.py:

This file contains zero characters. Opening it should show a completely blank editor.

app/main.py:

from fastapi import FastAPI

app = FastAPI(title="BeSure PRD Attachment Ingestion Prototype")


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}

What should match in the app?

The file contains one FastAPI application plus one /health route. Its response contains only the status value.

requirements.txt:

azure-ai-documentintelligence==1.0.2
azure-core==1.41.0
azure-storage-blob==12.30.2
fastapi==0.141.1
pydantic==2.13.5
python-multipart==0.0.32
uvicorn==0.53.0

What should match in the requirements?

Every package has an exact version pin. The file contains seven package entries.

settings.local.json.example:

{
  "document_intelligence_endpoint": "your-document-intelligence-endpoint",
  "document_intelligence_key": "your-document-intelligence-key",
  "storage_connection_string": "your-storage-connection-string",
  "output_container": "your-existing-output-container"
}

What should match in the template?

The template contains the four required configuration keys. Each key keeps a reusable placeholder value.

Your settings.local.json file contains these same four keys. Its values come from your existing Azure resources.

Before you start the server, ask yourself whether the health route can return its small JSON response.

• Return to the Visual Studio Code terminal from earlier.
• Start the local application on the explicit localhost address by running this command:

.\.venv\Scripts\python.exe -m uvicorn app.main:app --host 127.0.0.1 --port 8000

What does this command start?

The virtual environment interpreter starts Uvicorn with the FastAPI instance from app/main.py.

The server keeps control of this terminal while it runs. Press Ctrl+C when you need to stop it.

• Open http://127.0.0.1:8000/health in your browser.

You will see {"status":"ok"} in the browser. This response proves that the pinned environment can import and serve the local application.

Health page not loading?

Confirm that the Uvicorn command is still running in the Visual Studio Code terminal. Check that the browser address uses port 8000 plus the /health path.

If the terminal reports an import problem, compare app/main.py plus app/__init__.py with the reference tab.

Help me diagnose the local health endpoint.

That is the local foundation working. Your pinned Windows environment now serves the prototype health response.

Next, you will give this running service its first manual PRD attachment boundary.

Capture a Manual PRD Upload

Your health endpoint proved that the local FastAPI app can answer requests. A PRD still needs a clear way into the prototype.

The manual entry path uses a multipart upload. It converts the request into one AttachmentEnvelope that later processing can trust.

In this step, get ready to:

• Define one normalized attachment contract for every ingestion source.
• Add a manual upload route with PRD identity fields.
• Verify the first manual upload response.

Shape the attachment envelope

Each ingestion source needs to describe an attachment in the same way. The AttachmentEnvelope contract carries the document bytes alongside the identity details needed by the shared pipeline.

• Create app/models.py in the file sidebar for the existing prd-ingestion-prototype folder in Visual Studio Code.
• Build the normalized attachment contract by adding this code:

from dataclasses import dataclass


@dataclass(frozen=True)
class AttachmentEnvelope:
    source: str
    prd_id: str
    attachment_id: str
    filename: str
    content_type: str
    content: bytes

What does this contract do?

• The @dataclass(frozen=True) decorator creates an immutable attachment record.
• The identity fields preserve where the attachment came from.
• The content field keeps the uploaded file as bytes for later processing.

• Save app/models.py.
• Confirm that models.py is listed inside the app folder in the file sidebar.

The new file gives every ingestion route one stable shape for passing attachments into the application.

Cannot find the attachment model?

Check that models.py sits inside the app folder. A file beside the app folder uses a different import path.

Confirm that the filename ends with .py. Help me check why app.models cannot be imported.

✔️ Awesome, I've got everything!

Your attachment contract is saved inside the app package.

ⓧ I'd like to double check the full code

from dataclasses import dataclass


@dataclass(frozen=True)
class AttachmentEnvelope:
    source: str
    prd_id: str
    attachment_id: str
    filename: str
    content_type: str
    content: bytes

Wire the manual upload route

FastAPI represents the incoming document with UploadFile. Its file object gives the route direct access to the uploaded bytes.

• Return to app/main.py in the editor.
• Find this import at the top of the file:

from fastapi import FastAPI

Why does this import change?

The current import exposes the application class. The manual route also needs the upload type supplied by FastAPI.

• Replace the current import section with this code:

from fastapi import FastAPI, UploadFile

from app.models import AttachmentEnvelope

What is available now?

The route can now receive an uploaded file through UploadFile. It can also normalize that file with AttachmentEnvelope.

• Find the health() function in app/main.py.
• Add the manual upload route below the health function by pasting this code:

@app.post("/ingest/manual")
def ingest_manual(
    prd_id: str,
    attachment_id: str,
    file: UploadFile,
) -> dict[str, object]:
    content = file.file.read()
    attachment = AttachmentEnvelope(
        source="manual",
        prd_id=prd_id,
        attachment_id=attachment_id,
        filename=file.filename or "uploaded-document",
        content_type=file.content_type or "",
        content=content,
    )
    return {
        "source": attachment.source,
        "filename": attachment.filename,
        "byte_count": len(attachment.content),
    }

What does this route do?

• The route accepts a PRD identifier.
• The route accepts an attachment identifier.
• The file.file.read() call reads the uploaded document into bytes.
• The AttachmentEnvelope(...) call gives the uploaded document the shared in-memory shape.
• The returned dictionary makes the first ingestion result visible in the interactive documentation.

• Save app/main.py.
• Return to the terminal where the local server is running.
• Stop the current server by pressing Ctrl+C.
• Restart the local application by running this command:

.\.venv\Scripts\python.exe -m uvicorn app.main:app --host 127.0.0.1 --port 8000

What does this command start?

• The virtual environment interpreter runs Uvicorn with the project dependencies.
• The app.main:app value points the server to the FastAPI application.
• The host keeps the prototype on the local computer.
• The port makes the application available at http://127.0.0.1:8000.

The terminal remains occupied while the local server listens for requests.

Server stopped during startup?

Check that the terminal is inside the existing prd-ingestion-prototype folder. The command depends on the virtual environment stored there.

Compare the import names in app/main.py with the class name in app/models.py. Help me diagnose why the manual upload route prevents Uvicorn from starting.

✔️ Awesome, I've got everything!

Your manual upload route is saved. The local server is ready for a real PRD request.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, UploadFile

from app.models import AttachmentEnvelope

app = FastAPI(title="BeSure PRD Attachment Ingestion Prototype")


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/ingest/manual")
def ingest_manual(
    prd_id: str,
    attachment_id: str,
    file: UploadFile,
) -> dict[str, object]:
    content = file.file.read()
    attachment = AttachmentEnvelope(
        source="manual",
        prd_id=prd_id,
        attachment_id=attachment_id,
        filename=file.filename or "uploaded-document",
        content_type=file.content_type or "",
        content=content,
    )
    return {
        "source": attachment.source,
        "filename": attachment.filename,
        "byte_count": len(attachment.content),
    }

Submit a manual PRD

FastAPI generates interactive documentation from the route definition. The page renders request fields for the two identifiers plus the uploaded file.

• Open http://127.0.0.1:8000/docs in your browser.
• Expand POST /ingest/manual.
• Select the control that enables request editing.
• Enter manual-prd-001 for prd_id.
• Enter manual-attachment-001 for attachment_id.
• Attach a PRD file with the file picker.

Before you submit it, how far do you think this first route carries the document?

• Submit the request.

Your first PRD now reaches the normalized ingestion boundary.

You will see a JSON response containing source, filename, and byte_count. The source value is manual.

• Open the existing Azure Blob Storage view for the configured output container.
• Check the blob list for entries containing manual-prd-001 or manual-attachment-001.

The configured output container remains unchanged. This is the intended shortfall.

Ingestion currently stops after returning metadata. Markdown output is still absent.

Upload request did not complete?

Confirm that the local server terminal is still running. A stopped server leaves the documentation page unable to submit the request.

Make sure every request field contains a value. Help me diagnose why POST /ingest/manual did not return file metadata.

Your manual route can now capture a real PRD. Next, you will connect this envelope to the shared Azure processing pattern.

Reuse Document Intelligence and Blob Storage

Your manual FastAPI route now captures a PRD attachment inside one normalized envelope. Its response still stops at file metadata.

This step routes that envelope through Azure AI Document Intelligence for Markdown conversion. The same processor stores the source document plus its artifacts in Azure Blob Storage.

Why one shared processor?

One processor gives every ingestion source the same Azure calls. It also gives every source the same artifact paths.

Separate processors would duplicate storage rules. They would also make future extraction changes harder to apply consistently.

In this step, get ready to:

• Load the existing Azure configuration through a validated settings module.
• Build one processor for Document Intelligence conversion and Blob uploads.
• Send the manual attachment envelope through the shared processor.

Load and validate Azure settings

The processor needs four values from the existing settings.local.json file. A dedicated settings module checks those values before an Azure client uses them.

• Switch back to Visual Studio Code from the earlier step.
• Use Visual Studio Code to create config.py inside the app folder.
• Add the settings data structure to app/config.py by pasting this code:

import json
from dataclasses import dataclass
from pathlib import Path


@dataclass(frozen=True)
class Settings:
    document_intelligence_endpoint: str
    document_intelligence_key: str
    storage_connection_string: str
    output_container: str

What does this code do?

• The imports provide JSON reading plus a concrete file path.
• The Settings data class groups the four values needed by the Azure clients.
• The frozen setting prevents code from replacing configuration values after loading.

• Save app/config.py.
• Confirm the file shows all four settings fields.

Seeing a settings syntax warning?

Check that each field remains indented beneath Settings.

Ask for help with the file structure here: Help me check the Settings data class in app/config.py.

• Add configuration loading below the Settings class by pasting this code:

def load_settings(path: Path = Path("settings.local.json")) -> Settings:
    if not path.exists():
        raise RuntimeError(
            "settings.local.json is missing. Duplicate settings.local.json.example and add your Azure values."
        )

    values = json.loads(path.read_text(encoding="utf-8"))
    required = (
        "document_intelligence_endpoint",
        "document_intelligence_key",
        "storage_connection_string",
        "output_container",
    )
    missing = [name for name in required if not values.get(name)]
    if missing:
        raise RuntimeError(f"Missing settings: {', '.join(missing)}")

    return Settings(**{name: values[name] for name in required})

How does validation work?

• The file check stops startup when settings.local.json is unavailable.
• The required tuple defines the accepted configuration keys.
• The missing-value check reports every empty required setting in one error.
• The final line creates one validated Settings object.

• Save app/config.py.
• Confirm the file shows load_settings() below the Settings class.

Keep secrets local

The loader reads the existing local settings file at runtime. Your .gitignore already excludes that file from source control.

✔️ Awesome, I've got everything!

Your configuration module now validates the existing local Azure settings.

ⓧ I'd like to double check the full code

import json
from dataclasses import dataclass
from pathlib import Path


@dataclass(frozen=True)
class Settings:
    document_intelligence_endpoint: str
    document_intelligence_key: str
    storage_connection_string: str
    output_container: str


def load_settings(path: Path = Path("settings.local.json")) -> Settings:
    if not path.exists():
        raise RuntimeError(
            "settings.local.json is missing. Duplicate settings.local.json.example and add your Azure values."
        )

    values = json.loads(path.read_text(encoding="utf-8"))
    required = (
        "document_intelligence_endpoint",
        "document_intelligence_key",
        "storage_connection_string",
        "output_container",
    )
    missing = [name for name in required if not values.get(name)]
    if missing:
        raise RuntimeError(f"Missing settings: {', '.join(missing)}")

    return Settings(**{name: values[name] for name in required})

This reference shows the complete configuration module for comparison.

Build the shared Azure service

The service module owns artifact naming plus Azure communication. This keeps the manual route focused on creating an attachment envelope.

• Use Visual Studio Code to create services.py inside the app folder.
• Add the service imports plus safe path helpers to app/services.py by pasting this code:

import io
import json
import re
from datetime import datetime, timezone

from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.ai.documentintelligence.models import DocumentContentFormat
from azure.core.credentials import AzureKeyCredential
from azure.storage.blob import BlobServiceClient

from app.config import Settings
from app.models import AttachmentEnvelope


def _safe_segment(value: str) -> str:
    cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip(".-")
    return cleaned or "unknown"


def _safe_filename(value: str) -> str:
    leaf_name = value.replace("\\", "/").split("/")[-1]
    return _safe_segment(leaf_name)

What do these helpers protect?

• The Azure imports provide document analysis plus Blob access.
• The _safe_segment() helper replaces unsupported path characters with hyphens.
• The _safe_filename() helper removes any folder path from the uploaded filename.
• These helpers create predictable Blob prefixes from attachment metadata.

• Save app/services.py.
• Confirm the editor lists _safe_segment() plus _safe_filename() in the new file.

Seeing an import warning?

Confirm Visual Studio Code is using the .venv interpreter created earlier.

Ask for help with the selected interpreter here: Help me resolve the Azure import warnings in app/services.py.

• Add the Blob store plus processor constructor below _safe_filename() by pasting this code:

class AzureBlobStore:
    def __init__(self, connection_string: str):
        self.service = BlobServiceClient.from_connection_string(
            conn_str=connection_string
        )

    def upload_bytes(
        self,
        container: str,
        blob_name: str,
        content: bytes,
    ) -> None:
        blob_client = self.service.get_blob_client(
            container=container,
            blob=blob_name,
        )
        blob_client.upload_blob(data=content, overwrite=True)


class AttachmentProcessor:
    def __init__(self, settings: Settings, blob_store: AzureBlobStore):
        self.settings = settings
        self.blob_store = blob_store
        self.document_client = DocumentIntelligenceClient(
            endpoint=settings.document_intelligence_endpoint,
            credential=AzureKeyCredential(settings.document_intelligence_key),
        )

How do the Azure clients fit together?

• The AzureBlobStore wraps the existing Storage account connection.
• The upload_bytes() method writes bytes to a selected container path.
• The overwrite=True option supports repeatable tests with the same attachment identity.
• The AttachmentProcessor creates one Document Intelligence client from the validated settings.

• Save app/services.py.
• Confirm the file now contains AzureBlobStore plus AttachmentProcessor.

• In app/main.py locate this current import block:

from fastapi import FastAPI, UploadFile

from app.models import AttachmentEnvelope

What are you replacing?

This current block only imports the web application types plus the attachment envelope. The updated block also brings in configuration plus the shared service classes.

• Replace the current import block with this updated block:

from functools import lru_cache

from fastapi import FastAPI, UploadFile

from app.config import load_settings
from app.models import AttachmentEnvelope
from app.services import AttachmentProcessor, AzureBlobStore

Why add these imports?

The application now has access to the settings loader plus both service classes. The cache import supports one reusable processor instance.

• Add the processor factory below the app declaration by pasting this code:

@lru_cache
def get_processor() -> AttachmentProcessor:
    settings = load_settings()
    blob_store = AzureBlobStore(settings.storage_connection_string)
    return AttachmentProcessor(settings=settings, blob_store=blob_store)

Why cache the processor?

• The first call loads the existing local settings.
• The factory creates one Blob store plus one processor.
• The cache returns that processor for later manual requests.

• Save app/main.py.
• Restart the local FastAPI server by reusing the command from Step 1.
• Refresh the existing health page from earlier.

You should still see {"status":"ok"}. This confirms the application can import the new modules.

Server no longer starts?

Check the terminal for the first file and line named in the traceback. A missing comma or incorrect indentation in app/config.py or app/services.py can stop the import.

Ask for help with the startup traceback here: Help me diagnose why the FastAPI server stopped after adding the Azure service modules.

Process a manual upload end to end

The processor now needs a path for each attachment. It will use that path for the original document plus every generated artifact.

• Add the first part of process() inside AttachmentProcessor by pasting this code below __init__():

    def process(self, attachment: AttachmentEnvelope) -> dict[str, object]:
        prd_id = _safe_segment(attachment.prd_id)
        attachment_id = _safe_segment(attachment.attachment_id)
        filename = _safe_filename(attachment.filename)
        prefix = f"prd/{prd_id}/attachments/{attachment_id}"

        source_blob = f"{prefix}/source/{filename}"
        markdown_blob = f"{prefix}/processed/content.md"
        manifest_blob = f"{prefix}/manifest.json"

        self.blob_store.upload_bytes(
            self.settings.output_container,
            source_blob,
            attachment.content,
        )

How are artifact paths created?

• The sanitized PRD identity plus attachment identity form one shared prefix.
• The original filename sits beneath the source path.
• The generated Markdown gets a stable processed/content.md path.
• The source upload preserves the original attachment bytes before analysis.

• Save app/services.py.
• Restart the local FastAPI server by reusing the command from Step 1.
• Refresh the existing health page from earlier.

You should see {"status":"ok"} again. The updated processor method now imports without a syntax error.

• Add Document Intelligence analysis after the source upload by pasting this code:

        analyze_options: dict[str, object] = {}
        if attachment.content_type:
            analyze_options["content_type"] = attachment.content_type

        poller = self.document_client.begin_analyze_document(
            model_id="prebuilt-layout",
            body=io.BytesIO(attachment.content),
            output_content_format=DocumentContentFormat.MARKDOWN,
            **analyze_options,
        )
        result = poller.result()
        markdown = result.content or ""
        if not markdown.strip():
            raise RuntimeError("Document Intelligence returned empty Markdown content.")

How does the conversion work?

• The optional content type gives the analyzer information about the uploaded file.
• The prebuilt-layout model analyzes the attachment bytes.
• The DocumentContentFormat.MARKDOWN option requests Markdown output.
• The empty-content check prevents a successful manifest from describing unusable output.

• Save app/services.py.
• Confirm begin_analyze_document() sits after the source upload inside process().

Analysis block looks misaligned?

Keep the entire analysis block indented inside process(). Keep the arguments beneath begin_analyze_document() indented one level further.

Ask for help with the indentation here: Help me place the Document Intelligence analysis block inside process().

• Complete process() below the empty Markdown check by pasting this code:

        self.blob_store.upload_bytes(
            self.settings.output_container,
            markdown_blob,
            markdown.encode("utf-8"),
        )

        manifest: dict[str, object] = {
            "status": "processed",
            "source": attachment.source,
            "prd_id": attachment.prd_id,
            "attachment_id": attachment.attachment_id,
            "filename": attachment.filename,
            "content_type": attachment.content_type,
            "processed_at": datetime.now(timezone.utc).isoformat(),
            "artifacts": {
                "source": source_blob,
                "markdown": markdown_blob,
                "manifest": manifest_blob,
            },
        }
        self.blob_store.upload_bytes(
            self.settings.output_container,
            manifest_blob,
            json.dumps(manifest, indent=2).encode("utf-8"),
        )
        return manifest

What does the manifest prove?

• The first upload stores the generated Markdown bytes.
• The manifest records a processed status plus the original attachment metadata.
• The artifacts object records all three Blob paths.
• The final upload stores the manifest before returning the same data to the caller.

• Save app/services.py.
• Restart the local FastAPI server by reusing the command from Step 1.
• Refresh the existing health page from earlier.

You should see {"status":"ok"}. The complete processing service now loads successfully.

✔️ Awesome, I've got everything!

Your shared service can now preserve the source document plus its generated Markdown and manifest.

ⓧ I'd like to double check the full code

import io
import json
import re
from datetime import datetime, timezone

from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.ai.documentintelligence.models import DocumentContentFormat
from azure.core.credentials import AzureKeyCredential
from azure.storage.blob import BlobServiceClient

from app.config import Settings
from app.models import AttachmentEnvelope


def _safe_segment(value: str) -> str:
    cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip(".-")
    return cleaned or "unknown"


def _safe_filename(value: str) -> str:
    leaf_name = value.replace("\\", "/").split("/")[-1]
    return _safe_segment(leaf_name)


class AzureBlobStore:
    def __init__(self, connection_string: str):
        self.service = BlobServiceClient.from_connection_string(
            conn_str=connection_string
        )

    def upload_bytes(
        self,
        container: str,
        blob_name: str,
        content: bytes,
    ) -> None:
        blob_client = self.service.get_blob_client(
            container=container,
            blob=blob_name,
        )
        blob_client.upload_blob(data=content, overwrite=True)


class AttachmentProcessor:
    def __init__(self, settings: Settings, blob_store: AzureBlobStore):
        self.settings = settings
        self.blob_store = blob_store
        self.document_client = DocumentIntelligenceClient(
            endpoint=settings.document_intelligence_endpoint,
            credential=AzureKeyCredential(settings.document_intelligence_key),
        )

    def process(self, attachment: AttachmentEnvelope) -> dict[str, object]:
        prd_id = _safe_segment(attachment.prd_id)
        attachment_id = _safe_segment(attachment.attachment_id)
        filename = _safe_filename(attachment.filename)
        prefix = f"prd/{prd_id}/attachments/{attachment_id}"

        source_blob = f"{prefix}/source/{filename}"
        markdown_blob = f"{prefix}/processed/content.md"
        manifest_blob = f"{prefix}/manifest.json"

        self.blob_store.upload_bytes(
            self.settings.output_container,
            source_blob,
            attachment.content,
        )

        analyze_options: dict[str, object] = {}
        if attachment.content_type:
            analyze_options["content_type"] = attachment.content_type

        poller = self.document_client.begin_analyze_document(
            model_id="prebuilt-layout",
            body=io.BytesIO(attachment.content),
            output_content_format=DocumentContentFormat.MARKDOWN,
            **analyze_options,
        )
        result = poller.result()
        markdown = result.content or ""
        if not markdown.strip():
            raise RuntimeError("Document Intelligence returned empty Markdown content.")

        self.blob_store.upload_bytes(
            self.settings.output_container,
            markdown_blob,
            markdown.encode("utf-8"),
        )

        manifest: dict[str, object] = {
            "status": "processed",
            "source": attachment.source,
            "prd_id": attachment.prd_id,
            "attachment_id": attachment.attachment_id,
            "filename": attachment.filename,
            "content_type": attachment.content_type,
            "processed_at": datetime.now(timezone.utc).isoformat(),
            "artifacts": {
                "source": source_blob,
                "markdown": markdown_blob,
                "manifest": manifest_blob,
            },
        }
        self.blob_store.upload_bytes(
            self.settings.output_container,
            manifest_blob,
            json.dumps(manifest, indent=2).encode("utf-8"),
        )
        return manifest

This reference shows the complete shared service module for comparison.

• In app/main.py find the current metadata return block:

    return {
        "source": attachment.source,
        "filename": attachment.filename,
        "byte_count": len(attachment.content),
    }

What is the current gap?

This block ends the request after normalization. It never sends the envelope to the Azure processor.

• Replace the metadata return block with this processor call:

    return get_processor().process(attachment)

What changes at the route boundary?

The route keeps its existing upload logic. Its final line now passes the normalized envelope into the shared processor.

• Save app/main.py.
• Restart the local FastAPI server by reusing the command from Step 1.

✔️ Awesome, I've got everything!

The manual route now delegates document processing to the shared processor.

ⓧ I'd like to double check the full code

from functools import lru_cache

from fastapi import FastAPI, UploadFile

from app.config import load_settings
from app.models import AttachmentEnvelope
from app.services import AttachmentProcessor, AzureBlobStore

app = FastAPI(title="BeSure PRD Attachment Ingestion Prototype")


@lru_cache
def get_processor() -> AttachmentProcessor:
    settings = load_settings()
    blob_store = AzureBlobStore(settings.storage_connection_string)
    return AttachmentProcessor(settings=settings, blob_store=blob_store)


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/ingest/manual")
def ingest_manual(
    prd_id: str,
    attachment_id: str,
    file: UploadFile,
) -> dict[str, object]:
    content = file.file.read()
    attachment = AttachmentEnvelope(
        source="manual",
        prd_id=prd_id,
        attachment_id=attachment_id,
        filename=file.filename or "uploaded-document",
        content_type=file.content_type or "",
        content=content,
    )
    return get_processor().process(attachment)

This reference shows the complete manual ingestion application for comparison.

Your next upload calls the existing Azure resources. One small PRD keeps this billing-backed test controlled.

• Return to the interactive API documentation.
• Select the POST /ingest/manual operation.
• Enable the request form.
• Reuse the prd_id from your earlier manual upload.
• Reuse the attachment_id from your earlier manual upload.
• Select the same PRD file from your earlier test.

Before you submit the request, do you expect the response to stop at file metadata or report completed processing?

• Submit the manual ingestion request.

The request may stay in progress while Azure analyzes the file. That pause means the Layout operation is still running.

You should receive a manifest with status set to processed. Its artifacts object should list source plus Markdown and manifest paths.

• Open the existing output container with your usual Azure storage interface.
• Navigate to prd/{safe-prd-id}/attachments/{safe-attachment-id}/.
• Confirm source/{safe-filename} contains the uploaded document.
• Confirm processed/content.md contains generated Markdown.
• Confirm manifest.json records the processed attachment.

That closes the processing gap. Your manual upload now produces a reusable source copy plus Azure-generated Markdown and an observable manifest.

Manual processing failed?

Check that all four values in settings.local.json are present. Keep their values out of screenshots.

Confirm the existing output container accepts Blob uploads from the supplied Storage connection.

Ask for help with the first terminal error here: Help me diagnose my failed manual PRD processing request.

Manual attachments now reach one Azure processing pipeline. Next up, you will connect an API-triggered source Blob to that same processor.

Add API Ingestion from Blob

Your manual upload now turns a PRD into a source copy, generated Markdown, and a processed manifest.

BeSure also receives attachments that already exist in Azure Blob Storage. This step gives your FastAPI application a machine-triggered route without creating a second processing pipeline.

In this step, get ready to:

• Define the validated JSON request for an existing source blob.
• Download the source blob into memory as bytes.
• Send the normalized attachment through the unchanged shared processor.

Define the API request contract

A machine-triggered route needs a predictable request shape. A Pydantic model validates each source coordinate before your route tries to retrieve the document.

• In app/models.py, add the Pydantic import below the existing dataclass import by inserting this line:

from pydantic import BaseModel

Why Import BaseModel?

• BaseModel defines the request body contract used by the API route.
• FastAPI uses that contract to read and validate the incoming JSON body.

• Add the API request model at the end of app/models.py by pasting this class:

class ApiIngestionRequest(BaseModel):
    prd_id: str
    attachment_id: str
    source_container: str
    source_blob: str
    filename: str
    content_type: str

How Does the Request Contract Help?

• The identity fields determine the output artifact prefix.
• The source coordinates tell the storage adapter which existing blob to retrieve.
• The filename and content type preserve the metadata needed by the shared attachment envelope.

• Save app/models.py.
• Return to the terminal running Uvicorn.
• Stop the running process by pressing Ctrl+C.
• Restart the process with the same command from Step 1.
• Refresh http://127.0.0.1:8000/health in your browser.

You should see {"status":"ok"}. This confirms the application can import the new request model.

Server Not Restarting?

Check that from pydantic import BaseModel sits below the dataclass import. Confirm that ApiIngestionRequest starts at the left edge of app/models.py.

Ask for targeted help with the import failure: Help me find why app/models.py cannot import ApiIngestionRequest.

Add Blob download support

The API route starts with coordinates instead of uploaded bytes. The storage adapter therefore needs to copy the referenced blob into an in-memory byte stream.

• In app/services.py, locate the AzureBlobStore class.
• Add this method below __init__() and above upload_bytes() by pasting the code:

    def download_bytes(self, container: str, blob_name: str) -> bytes:
        blob_client = self.service.get_blob_client(
            container=container,
            blob=blob_name,
        )
        destination = io.BytesIO()
        blob_client.download_blob().readinto(destination)
        return destination.getvalue()

What Does This Method Do?

• The blob client targets the source container and blob named by the request.
• The in-memory stream receives the downloaded binary content.
• The returned bytes match the content type already expected by AttachmentEnvelope.

• Save app/services.py.
• Return to the Uvicorn terminal.
• Stop the running process by pressing Ctrl+C.
• Restart the process with the same command from Step 1.
• Refresh http://127.0.0.1:8000/health in your browser.

You should still see {"status":"ok"}. The storage adapter can now download bytes without changing the existing upload behavior.

Blob Download Method Causing a Startup Failure?

Confirm that download_bytes() is indented inside AzureBlobStore. Check that it appears before upload_bytes().

Ask for help checking the method placement: Help me debug my AzureBlobStore download_bytes method.

Route API requests through the shared processor

The final route acts as an ingestion adapter. It retrieves bytes, builds the same attachment envelope, and hands that envelope to the existing processor.

• In app/main.py, replace the import group at the top of the file with this updated group:

from functools import lru_cache

from fastapi import FastAPI, UploadFile

from app.config import load_settings
from app.models import ApiIngestionRequest, AttachmentEnvelope
from app.services import AttachmentProcessor, AzureBlobStore

Why Update the Import Group?

The new route needs ApiIngestionRequest for its validated request body. The existing manual route continues to use AttachmentEnvelope.

• Add the API ingestion route below ingest_manual() by pasting this function:

@app.post("/ingest/api")
def ingest_api(request: ApiIngestionRequest) -> dict[str, object]:
    processor = get_processor()
    content = processor.blob_store.download_bytes(
        container=request.source_container,
        blob_name=request.source_blob,
    )
    attachment = AttachmentEnvelope(
        source="api",
        prd_id=request.prd_id,
        attachment_id=request.attachment_id,
        filename=request.filename,
        content_type=request.content_type,
        content=content,
    )
    return processor.process(attachment)

Why Keep One Processor?

• The route retrieves the existing blob through the processor's shared storage adapter.
• The downloaded bytes become an AttachmentEnvelope with source set to api.
• The unchanged AttachmentProcessor.process() method creates the same source copy, Markdown file, and manifest shape used by manual uploads.

• Save app/main.py.

✔️ Awesome, I've got everything!

Your three updated files are ready. Make sure each file is saved before restarting the application.

ⓧ I'd like to double check the full code

Compare app/models.py with this complete version.

from dataclasses import dataclass

from pydantic import BaseModel


@dataclass(frozen=True)
class AttachmentEnvelope:
    source: str
    prd_id: str
    attachment_id: str
    filename: str
    content_type: str
    content: bytes


class ApiIngestionRequest(BaseModel):
    prd_id: str
    attachment_id: str
    source_container: str
    source_blob: str
    filename: str
    content_type: str

Compare app/services.py with this complete version.

import io
import json
import re
from datetime import datetime, timezone

from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.ai.documentintelligence.models import DocumentContentFormat
from azure.core.credentials import AzureKeyCredential
from azure.storage.blob import BlobServiceClient

from app.config import Settings
from app.models import AttachmentEnvelope


def _safe_segment(value: str) -> str:
    cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip(".-")
    return cleaned or "unknown"


def _safe_filename(value: str) -> str:
    leaf_name = value.replace("\\", "/").split("/")[-1]
    return _safe_segment(leaf_name)


class AzureBlobStore:
    def __init__(self, connection_string: str):
        self.service = BlobServiceClient.from_connection_string(
            conn_str=connection_string
        )

    def download_bytes(self, container: str, blob_name: str) -> bytes:
        blob_client = self.service.get_blob_client(
            container=container,
            blob=blob_name,
        )
        destination = io.BytesIO()
        blob_client.download_blob().readinto(destination)
        return destination.getvalue()

    def upload_bytes(
        self,
        container: str,
        blob_name: str,
        content: bytes,
    ) -> None:
        blob_client = self.service.get_blob_client(
            container=container,
            blob=blob_name,
        )
        blob_client.upload_blob(data=content, overwrite=True)


class AttachmentProcessor:
    def __init__(self, settings: Settings, blob_store: AzureBlobStore):
        self.settings = settings
        self.blob_store = blob_store
        self.document_client = DocumentIntelligenceClient(
            endpoint=settings.document_intelligence_endpoint,
            credential=AzureKeyCredential(settings.document_intelligence_key),
        )

    def process(self, attachment: AttachmentEnvelope) -> dict[str, object]:
        prd_id = _safe_segment(attachment.prd_id)
        attachment_id = _safe_segment(attachment.attachment_id)
        filename = _safe_filename(attachment.filename)
        prefix = f"prd/{prd_id}/attachments/{attachment_id}"

        source_blob = f"{prefix}/source/{filename}"
        markdown_blob = f"{prefix}/processed/content.md"
        manifest_blob = f"{prefix}/manifest.json"

        self.blob_store.upload_bytes(
            self.settings.output_container,
            source_blob,
            attachment.content,
        )

        analyze_options: dict[str, object] = {}
        if attachment.content_type:
            analyze_options["content_type"] = attachment.content_type

        poller = self.document_client.begin_analyze_document(
            model_id="prebuilt-layout",
            body=io.BytesIO(attachment.content),
            output_content_format=DocumentContentFormat.MARKDOWN,
            **analyze_options,
        )
        result = poller.result()
        markdown = result.content or ""
        if not markdown.strip():
            raise RuntimeError("Document Intelligence returned empty Markdown content.")

        self.blob_store.upload_bytes(
            self.settings.output_container,
            markdown_blob,
            markdown.encode("utf-8"),
        )

        manifest: dict[str, object] = {
            "status": "processed",
            "source": attachment.source,
            "prd_id": attachment.prd_id,
            "attachment_id": attachment.attachment_id,
            "filename": attachment.filename,
            "content_type": attachment.content_type,
            "processed_at": datetime.now(timezone.utc).isoformat(),
            "artifacts": {
                "source": source_blob,
                "markdown": markdown_blob,
                "manifest": manifest_blob,
            },
        }
        self.blob_store.upload_bytes(
            self.settings.output_container,
            manifest_blob,
            json.dumps(manifest, indent=2).encode("utf-8"),
        )
        return manifest

Compare app/main.py with this complete version.

from functools import lru_cache

from fastapi import FastAPI, UploadFile

from app.config import load_settings
from app.models import ApiIngestionRequest, AttachmentEnvelope
from app.services import AttachmentProcessor, AzureBlobStore

app = FastAPI(title="BeSure PRD Attachment Ingestion Prototype")


@lru_cache
def get_processor() -> AttachmentProcessor:
    settings = load_settings()
    blob_store = AzureBlobStore(settings.storage_connection_string)
    return AttachmentProcessor(settings=settings, blob_store=blob_store)


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/ingest/manual")
def ingest_manual(
    prd_id: str,
    attachment_id: str,
    file: UploadFile,
) -> dict[str, object]:
    content = file.file.read()
    attachment = AttachmentEnvelope(
        source="manual",
        prd_id=prd_id,
        attachment_id=attachment_id,
        filename=file.filename or "uploaded-document",
        content_type=file.content_type or "",
        content=content,
    )
    return get_processor().process(attachment)


@app.post("/ingest/api")
def ingest_api(request: ApiIngestionRequest) -> dict[str, object]:
    processor = get_processor()
    content = processor.blob_store.download_bytes(
        container=request.source_container,
        blob_name=request.source_blob,
    )
    attachment = AttachmentEnvelope(
        source="api",
        prd_id=request.prd_id,
        attachment_id=request.attachment_id,
        filename=request.filename,
        content_type=request.content_type,
        content=content,
    )
    return processor.process(attachment)

• Return to the Uvicorn terminal.
• Stop the running process by pressing Ctrl+C.
• Restart the process with the same command from Step 1.
• Return to http://127.0.0.1:8000/docs in your browser.

You should see POST /ingest/api alongside the manual route. Its request body contains all six fields from ApiIngestionRequest.

• Expand POST /ingest/api in the interactive API documentation.
• Enable editing for the request body.

• Set prd_id to api-prd-001.
• Set attachment_id to api-attachment-001.
• Set source_container to your existing source container.
• Set source_blob to path/to/existing-prd.pdf.
• Set filename to existing-prd.pdf.
• Set content_type to application/pdf.

Before you submit the request, predict whether its artifact paths will differ from those produced by the manual route.

• Submit the request from the endpoint panel.

The response should report status as processed. It should report source as api.

The artifacts object should list the same source, Markdown, and manifest keys produced by the manual route.

• Return to the existing output container you checked in Step 3.
• Open the prd prefix.
• Open the folder matching api-prd-001.
• Open the attachments prefix.
• Open the folder matching api-attachment-001.

You should see the original document under source. You should also see processed/content.md and manifest.json under the same attachment prefix.

• Open the new API manifest.json.
• Return to the manual manifest.json from Step 3.
• Compare the status value in both manifests.
• Compare the artifact key names in both manifests.

Both manifests use the same processed shape. Their source values identify which ingestion adapter supplied the shared envelope.

API Request Failing?

Confirm that the source container exists in the Storage account configured by storage_connection_string. Check that source_blob includes the blob's complete path inside that container.

Ask for help tracing the route without sharing your connection string: Help me debug why POST /ingest/api cannot process my existing source blob.

That is the convergence proved: manual uploads and existing source blobs now create the same artifact shape through one processor.

Next, you will isolate PRD field extraction so a future schema can change without touching either ingestion route.

Isolate the PRD Extraction Contract

Both ingestion routes now converge on one processor. Each route produces the same source copy plus Markdown in Azure Blob Storage.

The predefined PRD fields are still unconfirmed. A replaceable extraction contract keeps those unstable rules outside the ingestion routes.

In this step, get ready to:

• Define a replaceable PRD field extraction contract.
• Return an explicit deferred result without guessing fields.
• Persist the extraction result through the shared processor.

Define the extractor boundary

A Python Protocol describes the behavior an extractor must provide. The processor can depend on that behavior without knowing which extraction strategy sits behind it.

• Switch back to Visual Studio Code from earlier.
• Create an `extractor.py` file inside the existing `app` folder.
• Define the replaceable contract plus its deferred implementation by pasting this code into `app/extractor.py`:

from typing import Protocol


class PrdFieldExtractor(Protocol):
    def extract(self, markdown: str) -> dict[str, object]:
        ...


class DeferredPrdFieldExtractor:
    def extract(self, markdown: str) -> dict[str, object]:
        return {
            "status": "not_configured",
            "schema_version": None,
            "fields": {},
            "message": "PRD field extraction will be configured after the predefined fields are confirmed.",
            "markdown_character_count": len(markdown),
        }

What does this code do?

• `PrdFieldExtractor` defines one method that accepts Markdown and returns a structured dictionary.
• The ellipsis is a valid protocol method body that declares the required method shape.
• `DeferredPrdFieldExtractor` records that extraction is unconfigured while preserving useful processing metadata.
• `markdown_character_count` proves the extractor received the generated Markdown.

Why use a replaceable contract?

The processor only needs an object with an `extract()` method. A future implementation can apply the confirmed schema without changing either ingestion route.

This boundary keeps field decisions in one place. It protects the attachment pipeline from schema churn.

• Save `app/extractor.py`.
• Select `app/extractor.py` in the Explorer sidebar.

You should see `PrdFieldExtractor` followed by `DeferredPrdFieldExtractor` in the saved file.

Cannot find the extractor file?

• Confirm that `extractor.py` sits inside the existing `app` folder.
• Check that the filename uses the `.py` extension.

Ask for help checking the file location.

✔️ Awesome, I've got everything!

Your extraction boundary is ready. The shared processor can now receive a replaceable implementation.

ⓧ I'd like to double check the full code

from typing import Protocol


class PrdFieldExtractor(Protocol):
    def extract(self, markdown: str) -> dict[str, object]:
        ...


class DeferredPrdFieldExtractor:
    def extract(self, markdown: str) -> dict[str, object]:
        return {
            "status": "not_configured",
            "schema_version": None,
            "fields": {},
            "message": "PRD field extraction will be configured after the predefined fields are confirmed.",
            "markdown_character_count": len(markdown),
        }

How to use this reference

This reference shows the complete saved file. Compare its class names plus return keys with your version.

Inject the deferred extractor

Dependency injection gives the processor its extractor when the processor is created. This makes the extraction strategy explicit while preserving one shared processing path.

• In `app/services.py`, add this import directly below `from app.config import Settings`:

from app.extractor import PrdFieldExtractor

What does this import do?

The processor uses `PrdFieldExtractor` as its dependency type. Any implementation that follows this protocol can fill that role.

• In `app/services.py`, replace the existing `AttachmentProcessor` constructor with this version:

class AttachmentProcessor:
    def __init__(
        self,
        settings: Settings,
        blob_store: AzureBlobStore,
        extractor: PrdFieldExtractor,
    ):
        self.settings = settings
        self.blob_store = blob_store
        self.extractor = extractor
        self.document_client = DocumentIntelligenceClient(
            endpoint=settings.document_intelligence_endpoint,
            credential=AzureKeyCredential(settings.document_intelligence_key),
        )

What changes in the processor?

• The constructor requires an extractor that follows `PrdFieldExtractor`.
• `self.extractor` keeps that implementation available throughout processing.
• The existing settings plus Blob client remain unchanged.

• Save `app/services.py`.
• Confirm that `PrdFieldExtractor` appears in both the import group and the constructor.

• In `app/main.py`, add this import directly below `from app.config import load_settings`:

from app.extractor import DeferredPrdFieldExtractor

Why import the deferred implementation?

The application chooses `DeferredPrdFieldExtractor` as the current strategy. This keeps the schema gap explicit.

• In `app/main.py`, replace the existing `get_processor()` definition with this version:

@lru_cache
def get_processor() -> AttachmentProcessor:
    settings = load_settings()
    blob_store = AzureBlobStore(settings.storage_connection_string)
    return AttachmentProcessor(
        settings=settings,
        blob_store=blob_store,
        extractor=DeferredPrdFieldExtractor(),
    )

How does injection work here?

`get_processor()` constructs one deferred extractor and passes it into `AttachmentProcessor`. Both routes still receive the same cached processor.

• Save `app/main.py`.
• Return to the terminal running Uvicorn.
• Press `Ctrl+C` to stop the current process.
• Restart the application with the injected extractor by running this command:

.\.venv\Scripts\python.exe -m uvicorn app.main:app --host 127.0.0.1 --port 8000

What does this check prove?

Restarting loads the new imports plus constructor signature. The explicit host and port keep the prototype on the same local address.

• Return to the interactive API documentation at `http://127.0.0.1:8000/docs`.
• Submit one PRD through `POST /ingest/manual` with the same test values from earlier.

You should receive a processed response. This confirms that the manual route can construct the newly injected processor.

Does the request fail after restart?

• Check that `DeferredPrdFieldExtractor` is imported from `app.extractor` in `app/main.py`.
• Check that `extractor=DeferredPrdFieldExtractor()` appears inside the `AttachmentProcessor` call.
• Check that the constructor stores the dependency as `self.extractor`.

Ask for help tracing the injection error.

✔️ Awesome, I've got everything!

The application now injects one deferred extractor into the shared processor.

ⓧ I'd like to double check the full code

from functools import lru_cache

from fastapi import FastAPI, UploadFile

from app.config import load_settings
from app.extractor import DeferredPrdFieldExtractor
from app.models import ApiIngestionRequest, AttachmentEnvelope
from app.services import AttachmentProcessor, AzureBlobStore

app = FastAPI(title="BeSure PRD Attachment Ingestion Prototype")


@lru_cache
def get_processor() -> AttachmentProcessor:
    settings = load_settings()
    blob_store = AzureBlobStore(settings.storage_connection_string)
    return AttachmentProcessor(
        settings=settings,
        blob_store=blob_store,
        extractor=DeferredPrdFieldExtractor(),
    )


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/ingest/manual")
def ingest_manual(
    prd_id: str,
    attachment_id: str,
    file: UploadFile,
) -> dict[str, object]:
    content = file.file.read()
    attachment = AttachmentEnvelope(
        source="manual",
        prd_id=prd_id,
        attachment_id=attachment_id,
        filename=file.filename or "uploaded-document",
        content_type=file.content_type or "",
        content=content,
    )
    return get_processor().process(attachment)


@app.post("/ingest/api")
def ingest_api(request: ApiIngestionRequest) -> dict[str, object]:
    processor = get_processor()
    content = processor.blob_store.download_bytes(
        container=request.source_container,
        blob_name=request.source_blob,
    )
    attachment = AttachmentEnvelope(
        source="api",
        prd_id=request.prd_id,
        attachment_id=request.attachment_id,
        filename=request.filename,
        content_type=request.content_type,
        content=content,
    )
    return processor.process(attachment)

How to use this reference

This file keeps both route handlers unchanged. Only the extractor import plus processor construction are new.

Persist the deferred extraction

The extractor result becomes useful when it is stored beside the generated Markdown. The manifest also needs to report that result so callers can inspect processing state without opening every artifact.

• In `app/services.py`, locate the artifact path declarations directly below `prefix`.
• Replace that declaration group with this version:

        source_blob = f"{prefix}/source/{filename}"
        markdown_blob = f"{prefix}/processed/content.md"
        extraction_blob = f"{prefix}/processed/extraction.json"
        manifest_blob = f"{prefix}/manifest.json"

What changes in the artifact layout?

`extraction_blob` places structured extraction output beside `content.md`. Both ingestion routes inherit this path because they share `AttachmentProcessor.process()`.

• Locate the Markdown upload immediately after the empty Markdown check.
• Replace that upload section with this extraction plus persistence block:

        extraction = self.extractor.extract(markdown)

        self.blob_store.upload_bytes(
            self.settings.output_container,
            markdown_blob,
            markdown.encode("utf-8"),
        )
        self.blob_store.upload_bytes(
            self.settings.output_container,
            extraction_blob,
            json.dumps(extraction, indent=2).encode("utf-8"),
        )

What does this processing block do?

• `self.extractor.extract(markdown)` passes the generated Markdown through the injected contract.
• The Markdown remains stored as UTF-8 bytes in `content.md`.
• The extraction dictionary is serialized as formatted JSON in `extraction.json`.

• Save `app/services.py`.
• Return to the terminal running Uvicorn.
• Press `Ctrl+C` to stop the current process.
• Restart the application with extraction persistence by running this command:

.\.venv\Scripts\python.exe -m uvicorn app.main:app --host 127.0.0.1 --port 8000

Why restart again?

The running process still holds the previous `AttachmentProcessor` code. Restarting loads the extraction path plus upload stage.

• Return to `http://127.0.0.1:8000/docs`.
• Submit one PRD through `POST /ingest/manual`.
• Return to the configured output container from earlier.
• Inspect the attachment prefix created by the manual request.

You should see `processed/extraction.json` beside `processed/content.md`. The extraction file should contain `status` set to `not_configured`.

Is extraction.json missing?

• Confirm that `extraction_blob` ends with `processed/extraction.json`.
• Confirm that `self.extractor.extract(markdown)` appears before the Blob uploads.
• Confirm that the extraction upload uses `json.dumps(extraction, indent=2).encode("utf-8")`.

Ask for help tracing the missing artifact.

• In `app/services.py`, locate the existing `manifest` dictionary.
• Replace the complete dictionary with this version:

        manifest: dict[str, object] = {
            "status": "processed",
            "source": attachment.source,
            "prd_id": attachment.prd_id,
            "attachment_id": attachment.attachment_id,
            "filename": attachment.filename,
            "content_type": attachment.content_type,
            "processed_at": datetime.now(timezone.utc).isoformat(),
            "extraction_status": extraction["status"],
            "artifacts": {
                "source": source_blob,
                "markdown": markdown_blob,
                "extraction": extraction_blob,
                "manifest": manifest_blob,
            },
        }

How does the manifest expose extraction?

`extraction_status` surfaces the extractor result in the main response. The `extraction` artifact entry tells callers where the full JSON result is stored.

• Save `app/services.py`.
• Return to the terminal running Uvicorn.
• Press `Ctrl+C` to stop the current process.
• Restart the completed processor by running this command:

.\.venv\Scripts\python.exe -m uvicorn app.main:app --host 127.0.0.1 --port 8000

What is loaded now?

The restarted processor now runs the extractor. It stores `extraction.json` and reports the extraction state in `manifest.json`.

Existing Azure billing applies

These checks use the existing Azure resources from earlier. The calls plus retained test blobs can contribute to the account's normal Azure charges.

You are creating no new Azure resource in this step.

Before you test both paths, do you expect their extraction status and artifact names to match?

• Return to `http://127.0.0.1:8000/docs`.
• Process one document through `POST /ingest/manual`.
• Process one document through `POST /ingest/api` with valid source Blob coordinates.
• Inspect the completed manual attachment prefix in the configured output container.
• Inspect the completed API attachment prefix in the configured output container.

Both prefixes should contain the source document plus `processed/content.md`. They should also contain `processed/extraction.json` plus `manifest.json`.

Open each `extraction.json` file. You should see `status` set to `not_configured` plus `schema_version` set to `null`.

Open each `manifest.json` file. You should see `extraction_status` set to `not_configured` plus an `extraction` artifact path.

Do the two prefixes differ?

• Confirm that both routes still end with `processor.process(attachment)`.
• Confirm that the API route creates an `AttachmentEnvelope` before calling the processor.
• Confirm that the manual route still calls the cached processor from `get_processor()`.

Ask for help comparing the two processing paths.

✔️ Awesome, I've got everything!

Your shared processor now persists the same deferred extraction result for both ingestion routes.

ⓧ I'd like to double check the full code

import io
import json
import re
from datetime import datetime, timezone

from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.ai.documentintelligence.models import DocumentContentFormat
from azure.core.credentials import AzureKeyCredential
from azure.storage.blob import BlobServiceClient

from app.config import Settings
from app.extractor import PrdFieldExtractor
from app.models import AttachmentEnvelope


def _safe_segment(value: str) -> str:
    cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip(".-")
    return cleaned or "unknown"


def _safe_filename(value: str) -> str:
    leaf_name = value.replace("\\", "/").split("/")[-1]
    return _safe_segment(leaf_name)


class AzureBlobStore:
    def __init__(self, connection_string: str):
        self.service = BlobServiceClient.from_connection_string(
            conn_str=connection_string
        )

    def download_bytes(self, container: str, blob_name: str) -> bytes:
        blob_client = self.service.get_blob_client(
            container=container,
            blob=blob_name,
        )
        destination = io.BytesIO()
        blob_client.download_blob().readinto(destination)
        return destination.getvalue()

    def upload_bytes(
        self,
        container: str,
        blob_name: str,
        content: bytes,
    ) -> None:
        blob_client = self.service.get_blob_client(
            container=container,
            blob=blob_name,
        )
        blob_client.upload_blob(data=content, overwrite=True)


class AttachmentProcessor:
    def __init__(
        self,
        settings: Settings,
        blob_store: AzureBlobStore,
        extractor: PrdFieldExtractor,
    ):
        self.settings = settings
        self.blob_store = blob_store
        self.extractor = extractor
        self.document_client = DocumentIntelligenceClient(
            endpoint=settings.document_intelligence_endpoint,
            credential=AzureKeyCredential(settings.document_intelligence_key),
        )

    def process(self, attachment: AttachmentEnvelope) -> dict[str, object]:
        prd_id = _safe_segment(attachment.prd_id)
        attachment_id = _safe_segment(attachment.attachment_id)
        filename = _safe_filename(attachment.filename)
        prefix = f"prd/{prd_id}/attachments/{attachment_id}"

        source_blob = f"{prefix}/source/{filename}"
        markdown_blob = f"{prefix}/processed/content.md"
        extraction_blob = f"{prefix}/processed/extraction.json"
        manifest_blob = f"{prefix}/manifest.json"

        self.blob_store.upload_bytes(
            self.settings.output_container,
            source_blob,
            attachment.content,
        )

        analyze_options: dict[str, object] = {}
        if attachment.content_type:
            analyze_options["content_type"] = attachment.content_type

        poller = self.document_client.begin_analyze_document(
            model_id="prebuilt-layout",
            body=io.BytesIO(attachment.content),
            output_content_format=DocumentContentFormat.MARKDOWN,
            **analyze_options,
        )
        result = poller.result()
        markdown = result.content or ""
        if not markdown.strip():
            raise RuntimeError("Document Intelligence returned empty Markdown content.")

        extraction = self.extractor.extract(markdown)

        self.blob_store.upload_bytes(
            self.settings.output_container,
            markdown_blob,
            markdown.encode("utf-8"),
        )
        self.blob_store.upload_bytes(
            self.settings.output_container,
            extraction_blob,
            json.dumps(extraction, indent=2).encode("utf-8"),
        )

        manifest: dict[str, object] = {
            "status": "processed",
            "source": attachment.source,
            "prd_id": attachment.prd_id,
            "attachment_id": attachment.attachment_id,
            "filename": attachment.filename,
            "content_type": attachment.content_type,
            "processed_at": datetime.now(timezone.utc).isoformat(),
            "extraction_status": extraction["status"],
            "artifacts": {
                "source": source_blob,
                "markdown": markdown_blob,
                "extraction": extraction_blob,
                "manifest": manifest_blob,
            },
        }
        self.blob_store.upload_bytes(
            self.settings.output_container,
            manifest_blob,
            json.dumps(manifest, indent=2).encode("utf-8"),
        )
        return manifest

How to use this reference

This is the complete cumulative `app/services.py` file for this step. Compare the extractor import plus constructor dependency first.

Then compare the extraction upload plus manifest fields. Those sections create the new Blob artifact shape.

That completes the ingestion boundary. Both entry paths now produce the same deferred extraction artifact without committing BeSure to guessed PRD fields.

Secret mission

Persist Failure Manifests

A processing error currently reaches the API caller without leaving a durable record for operators. Extend the shared processor so every downstream failure writes a safe manifest while preserving the original exception.

Clean Up Your Resources

Clean Up Your Resources

No new Azure resources were created, though calls to Azure AI Document Intelligence plus retained test blobs in Azure Blob Storage use the existing account's normal billing configuration. Choose whether to keep the prototype, pause its local server, or delete its test data.

Cost warning

Processing calls can contribute to existing Azure charges. Blob operations plus retained test artifacts can also contribute to those charges.

• Choose the pause option if you plan to continue integration work later.
• Choose the delete option when the prototype test artifacts are no longer needed.

Resources you used:

• The local Uvicorn process serving your FastAPI application.
• The .venv virtual environment inside your prd-ingestion-prototype folder.
• The settings.local.json file inside your prd-ingestion-prototype folder. This file contains your existing Azure configuration values.
• Prototype Blob artifacts beneath each prd/{safe-prd-id}/attachments/{safe-attachment-id}/ prefix in the existing container named by output_container.

Keep everything running

This option suits active integration work. No cleanup is required.

• Retain each test Blob prefix for comparisons during integration.
• Retain .venv so the pinned dependencies remain available.
• Store settings.local.json only on this computer.
• Leave Uvicorn running only while you are actively testing the ingestion routes.

Keep Local Credentials Private

Your existing .gitignore excludes settings.local.json from source control. Keep that exclusion in place while retaining the prototype.

Pause - I'll come back to this later

This option stops the local API while preserving the prototype for later. Your existing Azure resources remain unchanged.

• Return to the terminal running Uvicorn.
• Press Ctrl+C to stop the local server.
• Retain .venv for the next development session.
• Retain settings.local.json on this computer.
• Retain the test Blob prefixes if you still need them for integration comparisons.

Your terminal returns to its command prompt. The local API is now safely offline.

Delete - I don't want to use this again

This option removes the prototype's test artifacts from the existing output container. It also removes the local virtual environment plus local Azure configuration.

Keep Existing Resources Safe

Deleting test data can feel risky. This cleanup targets only prefixes created during prototype testing.

Your existing containers stay in place. Your Azure resources also stay in place.

• Return to the terminal running Uvicorn.
• Press Ctrl+C to stop the local server.

Your local API is now offline. You can remove its stored test artifacts without another request arriving.

• In the Azure portal, open the existing storage account referenced by storage_connection_string.
• Open the existing container named by output_container.
• Navigate to the test prd/{safe-prd-id}/attachments/{safe-attachment-id}/ prefix.
• Select every blob stored beneath that test prefix.
• Click Delete in the container toolbar.
• Confirm the deletion when the portal asks.
• Repeat the prefix deletion for every test PRD attachment created during this project.
• Refresh the container view.

You should no longer see the prototype test prefixes. Your existing output container remains available.

• In File Explorer, open your prd-ingestion-prototype folder.
• Delete the .venv folder.
• Delete the settings.local.json file.
• Refresh File Explorer.

You should see that both local items are gone. The prototype source files remain available for integration work.

Nice Work!

Nice Work!

You did it! Your local FastAPI prototype now accepts PRD attachments through manual upload or API-triggered Blob retrieval.

Both entry paths converge on one processing pipeline. Each successful run leaves a consistent set of artifacts for inspection.

You've learned how to:

• Built a manual multipart upload path that converts each PRD attachment into a shared AttachmentEnvelope.
• Built an API-triggered ingestion path that retrieves an existing source blob. ApiIngestionRequest validates its JSON coordinates before the attachment enters the shared processor.
• Reused one shared processor to convert documents into Markdown with Azure AI Document Intelligence Layout. The processor writes consistent artifacts to Azure Blob Storage. The PrdFieldExtractor boundary keeps future field rules isolated from ingestion.
• Secret Mission: Added observable failure manifests for downstream processing exceptions. The shared processor records failed before re-raising the original exception to the API caller.

Ready to quiz yourself?