FastAPI and ORM Examples

[juftin]

TLDR; If you're using fhir.resources alongside FastAPI and/or an ORM like SQLAlchemy / Beanie, share your examples and best practice recommendations

---

By leveraging pydantic, this project was my go-to when trying to build out a micro-service that could send/receive FHIR messages using FastAPI.

Beanie was also an easy choice for the ORM (ODM in this case, object-document-mapper) since it leverages Pydantic alongside a NoSQL database, MongoDB.

However, when getting down into the implementation details I faced compatibility issues with fhir.resources.DomainResource instances with these other libraries. The customization of the pydantic code under the hood is what made this difficult.

I've run into issues with FastAPI response models showing internal fields with underscores, pydantic validation issues when using fhir.resources.DomainResource as request models, and issues with being able to make all of this persist to the NoSQL layer with Beanie.

I've created some patterns and parent classes to make all this work, not sure how hacky the community will find them to be. I'll share them below as a comment but wanted to reach out to the community to see if anyone else has any useful code snippets to share?

[juftin]

Here's an example implementation I built using pydantic, beanie, and fhir.resources:
https://github.com/juftin/fastapi-fhir/blob/ad784f6a119c49bd11d04a979f370b273814e8ea/fastapi_fhir/organization.py

And here is a quick-start doc for running MongoDB and the example app locally: https://github.com/juftin/fastapi-fhir/tree/ad784f6a119c49bd11d04a979f370b273814e8ea#quick-start

And here is the important bits of the example. In it you'll see that I create two for classes for use. The two classes can convert back and forth from eachother using the .from_orm() method:

• FhirOrganization - this is the fhir.resources model. It contains all of the correct code so that calling .dict() / .json() on these instances will produce a reliable HL7 FHIR message
• Organization - this is the beanie Document class representing the FhirOrganization as a document in MongoDB.

"""
Organization

https://www.hl7.org/fhir/organization.html
"""

from beanie import Document, init_beanie
from fastapi import FastAPI
from fastapi.responses import Response
from fhir.resources.R4B.organization import Organization as _FhirOrganization
from motor.motor_asyncio import AsyncIOMotorClient
from pydantic import BaseModel


class FhirOrganization(_FhirOrganization):
    """
    fhir.resources: Organization
    """

    class Config:
        """
        Model configuration
        """

        orm_mode = True


class Organization(FhirOrganization, Document):
    """
    fhir.resources: Organization + MongoDB Document
    """


app = FastAPI()


@app.on_event("startup")
async def startup_event():
    """
    Startup event

    Initialize the database on startup
    """
    client = AsyncIOMotorClient(
        "mongodb://localhost:27017",
    )
    database = client.get_default_database("fhir")
    await init_beanie(database=database, document_models=[Organization])


@app.get("/api/FHIR/R4/Organization/{id}", response_model=FhirOrganization)
async def get_organization(id: str) -> Response:
    """
    Get an Organization by ID

    This method returns a Response so that additional validation
    is not applied to the response. `Response` is used instead of
    `JSONResponse` to limit the number of times the response model
    is processed to JSON. fhir.resources customizes the
    pydantic model enough that FastAPI's validation fails if validation
    is attempted on the response model.
    """
    org = await Organization.find_one(Organization.id == id)
    if org is not None:
        fhir_org = FhirOrganization.from_orm(org)
        return Response(
            fhir_org.json(),
            media_type="application/json",
            status_code=200,
        )