I want to put different files, how to use backlink?

[hgalytoby]

Say I want to put 'House' and 'Door' in two different files, namely "house.py" and "door.py" respectively.

Let's presume door.py and house.py are neighbour to each others under the same parent folder, how do I implement backlinks?

main.py

import asyncio
from beanie import init_beanie
from motor import motor_asyncio

from house import House
from door import Door


async def init():
    client = motor_asyncio.AsyncIOMotorClient(
        'mongodb://root:12345678@localhost:27017'
    )
    await init_beanie(
        database=client.Noah,
        document_models=[
            House, Door,
        ],
    )


async def main():
    await init()
    print(await House.find().to_list())
    print(await Door.find().to_list())



if __name__ == '__main__':
    asyncio.run(main())

house.py

from beanie import Document, Link


class House(Document):
    name: str
    door: Link["Door"]

    class Settings:
        name = 'BackLinkHouse'

door.py

from beanie import Document, BackLink
from pydantic import Field

from house import House


class Door(Document):
    height: int = 2
    width: int = 1
    house: BackLink[House] = Field(original_field="door")

    class Settings:
        name = 'BackLinkDoor'

raise NameError: name 'Door' is not defined

If you change door: door: Link["Door"] to door: Link[Door] and from door import Door, a circular import will occur.

[hgalytoby]

I've found that doing this does the trick.

main.py

import asyncio
from beanie import init_beanie
from motor import motor_asyncio

from house import House
from door import Door

House.update_forward_refs(Door=Door)
Door.update_forward_refs(House=House)


async def init():
    client = motor_asyncio.AsyncIOMotorClient(
        'mongodb://root:12345678@localhost:27017'
    )
    await init_beanie(
        database=client.Noah,
        document_models=[
            House, Door,
        ],
    )


async def main():
    await init()
    print(await House.find().to_list())
    print(await Door.find().to_list())



if __name__ == '__main__':
    asyncio.run(main())

house.py

from beanie import Document, Link


class House(Document):
    name: str
    door: Link['Door']

    class Settings:
        name = 'BackLinkHouse'

door.py

from beanie import Document, BackLink
from pydantic import Field


class Door(Document):
    height: int = 2
    width: int = 1
    house: BackLink['House'] = Field(original_field="door")

    class Settings:
        name = 'BackLinkDoor'

Is there any other way?

[roman-right]

Your workaround is nice.

You can do things like this also:

File a_circular_import.py:

from __future__ import annotations

from typing import Optional

from pydantic import BaseModel


class A(BaseModel):
    b: Optional[B] = None


from b_circular_import import B

A.update_forward_refs()

File b_circular_import.py:

from __future__ import annotations

from pydantic import BaseModel


class B(BaseModel):
    a: A


from a_circular_import import A

B.update_forward_refs()

b = B(a=A())

print(
    b.json(indent=2)
)

But I like your solution more

[hgalytoby]

Okay, I understand. Thank you for your reply.

[diegocgaona]

Hi,

I'm trying to solve a similar problem.
I tried the solution from @hgalytoby with update_forward_refs , but it doesn't work anymore:
/pydantic/main.py", line 1311, in update_forward_refs raise TypeError('localns arguments are not longer accepted.')

And it seems that update_forward_refs is deprecated.

The solution from @roman-right give me this error when I try to access the API documentation: (I'm using FastAPI)

pydantic.errors.PydanticInvalidForJsonSchema: Cannot generate a JsonSchema for core_schema.PlainValidatorFunctionSchema ({'type': 'with-info', 'function': <function BackLink.build_validation.<locals>.validate at 0x7fa8308df240>})

And I'm using Document instead of BaseModel.

I tried some other things for hours, like forward_refs, some solutions for circular import, but I can't make it work :(

Here are some examples of my models which give me the problems:

class Ingredients(Document):
    id: PydanticObjectId = Field(default_factory=PydanticObjectId, alias="_id")
    name: Indexed(str, unique=True)
    used_in_drinks: List[BackLink[Drinks]] = Field(original_field="ingredients")
    ingredients_tags: Optional[list[IngredientsTags]] = None

    class Settings:
        name = "ingredients"
        keep_nulls = False

class Drinks(Document):
    id: PydanticObjectId = Field(default_factory=PydanticObjectId, alias="_id")
    name: Indexed(str, unique=True)
    ingredients: Optional[List[Link[Ingredients]]] = None
    drinks_tags: Optional[list[DrinksTags]]

    class Settings:
        name = "drinks"
        keep_nulls = False

Can someone have an idea how to solve it?

Best regards, Diego

[hgalytoby]

There is no need to use update_forward_refs, I tested it without problems.

pydantic 2.7

drinks.py

from typing import Optional, TYPE_CHECKING

from beanie import Document, PydanticObjectId, Indexed, Link
from pydantic import Field

if TYPE_CHECKING:
    from .ingredients import Ingredients


class Drinks(Document):
    id: PydanticObjectId = Field(default_factory=PydanticObjectId, alias="_id")
    name: Indexed(str, unique=True)
    ingredients: Optional[list[Link['Ingredients']]] = None

    class Settings:
        name = "drinks"
        keep_nulls = False

Ingredients.py

from beanie import Document, Indexed, PydanticObjectId, BackLink
from pydantic import Field
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from .drinks import Drinks


class Ingredients(Document):
    id: PydanticObjectId = Field(default_factory=PydanticObjectId, alias="_id")
    name: Indexed(str, unique=True)
    used_in_drinks: list[BackLink['Drinks']] = Field(original_field="ingredients")

    class Settings:
        name = "ingredients"
        keep_nulls = False

my_test.py

async def my_test():
    await init_beanie(
        database=db,
        document_models=[Drinks, Ingredients],
    )
    ingredients = Ingredients(name=uuid4().hex)
    await ingredients.insert()
    drinks = Drinks(name=uuid4().hex, ingredients=[ingredients])
    await drinks.insert()

[diegocgaona]

Hi @hgalytoby,

I think my problem is related to Pydantic generation of the json schema for the FastAPI OpenAPI documentation (it is what give me the error, the insert operation would work).
And it is related to the BackLink. If I change BackLink to Link, it works.

I made a minimum example bellow (you can try in a unique file and the errors happen same way, but uncomment the imports and you can test with the directory structure):

# database/ingredients_models.py
from pydantic import Field
from beanie import Document, BackLink
from typing import List
# from database.drinks_models import Drinks


class Ingredients(Document):
    name: str
    used_in_drinks: List[BackLink["Drinks"]] = Field(original_field="ingredients")


# database/drinks_models.py
from beanie import Document, Link
from typing import Optional
# from database.ingredients_models import Ingredients


class Drinks(Document):
    name: str
    ingredients: Optional[List[Link[Ingredients]]] = None


# database/db_config.py
import motor.motor_asyncio
import os

MONGODB_URL = os.getenv("MONGODB_URL", "")
client = motor.motor_asyncio.AsyncIOMotorClient(
    "MONGODB_URL", serverSelectionTimeoutMS=5000,
    uuidRepresentation="standard"
)

# routers/ingredients_router.py
from fastapi import APIRouter
# from database.ingredients_models import Ingredients

router = APIRouter(
    prefix="/ingredients",
    tags=["Ingredients"],
    responses={404: {"description": "Not Found"}},
)

response_schema = Ingredients


@router.get("/{item_id}", response_model=response_schema)
async def read_item(
        item: response_schema,
):
    return item  # Will return the item from the Depends


# main.py
from fastapi import FastAPI
import os
import uvicorn

from beanie import init_beanie
from contextlib import asynccontextmanager

from fastapi import FastAPI
# from database.db_config import client

# from database.drinks_models import Drinks
# from database.ingredients_models import Ingredients

# from routers import ingredients_router


async def my_test():
    await init_beanie(database=client.drinks,
                      document_models=[Drinks, Ingredients])
    yield


app = FastAPI(
    title="API Test",
    description="Test API",
    version="0.1.0",
)

# All the routers included
app.include_router(router)

if __name__ == "__main__":
    uvicorn.run("main:app", host="localhost",
                port=8080, reload=True, log_level="debug")

The error it gives me is:

raise PydanticInvalidForJsonSchema(f'Cannot generate a JsonSchema for {error_info}')
pydantic.errors.PydanticInvalidForJsonSchema: Cannot generate a JsonSchema for core_schema.PlainValidatorFunctionSchema ({'type': 'with-info', 'function': <function BackLink.build_validation.<locals>.validate at 0x7f7dee187560>})

@roman-right you think it could be a bug?

My packages are:
python = "^3.12"
fastapi = "^0.110.1"
beanie = "^1.25.0"
fastapi-users = { extras = ["beanie", "oauth"], version = "^13.0.0" }
uvicorn = "^0.29.0"
itsdangerous = "^2.1.2"
pydantic = "2.6.4"

Best regards, Diego.

[hgalytoby]

If you don't have fetch_links=True

Your field is a Link or BackLink will be an object that cannot be JsonSchema'ed.

exmaple

print(await Ingredients.find_all().to_list())
print(await Drinks.find_all().to_list())

output

[Ingredients(id=ObjectId('661959bf8925902b2b2b318b'), revision_id=None, name='08c24c38b9a74d36ac821c10887f7f46', used_in_drinks=[<beanie.odm.fields.BackLink object at 0x00000237D317CEE0>])]

[Drinks(id=ObjectId('661959bf8925902b2b2b318c'), revision_id=None, name='d183e1f458bf44caba52fcdf452d056e', ingredients=[<beanie.odm.fields.Link object at 0x00000237D30611E0>])]

If you use fetch_links=True then the data will look like this

exmaple

await Ingredients.find_all(fetch_links=True).to_list()

output

[Ingredients(id=ObjectId('661959bf8925902b2b2b318b'), revision_id=None, name='08c24c38b9a74d36ac821c10887f7f46', used_in_drinks=[Drinks(id=ObjectId('661959bf8925902b2b2b318c'), revision_id=None, name='d183e1f458bf44caba52fcdf452d056e', ingredients=[Ingredients(id=ObjectId('661959bf8925902b2b2b318b'), revision_id=None, name='08c24c38b9a74d36ac821c10887f7f46', used_in_drinks=[Drinks(id=ObjectId('661959bf8925902b2b2b318c'), revision_id=None, name='d183e1f458bf44caba52fcdf452d056e', ingredients=[<beanie.odm.fields.Link object at 0x000001A62735F1F0>])])])])]

To avoid duplicate access to data, you can limit the depth by referring to the documentation.

https://beanie-odm.dev/tutorial/defining-a-document/#nested-documents-depth

https://beanie-odm.dev/tutorial/relations/#nested-links

example

await Ingredients.find_all(fetch_links=True, nesting_depth=1).to_list()

output

[Ingredients(id=ObjectId('661959bf8925902b2b2b318b'), revision_id=None, name='08c24c38b9a74d36ac821c10887f7f46', used_in_drinks=[Drinks(id=ObjectId('661959bf8925902b2b2b318c'), revision_id=None, name='d183e1f458bf44caba52fcdf452d056e', ingredients=[<beanie.odm.fields.Link object at 0x000001FE6E0DFA90>])])]

Now the data problem is solved.

Next is the fastapi problem.

Since it can't be used by JsonSchema, you have to create an additional Response object.

class IngredientsResponse(BaseModel):
    id: PydanticObjectId = Field(alias='_id')
    name: str
    used_in_drinks: List["IngredientsDrinksResponse"] = Field(default_factory=list)


class IngredientsDrinksResponse(BaseModel):
    id: PydanticObjectId = Field(alias='_id')
    name: str


@router.get("/ingredients", response_model_by_alias=False)
async def ingredients_items() -> list[IngredientsResponse]:
    items = await Ingredients.find_all(fetch_links=True, projection_model=IngredientsResponse, nesting_depth=1).to_list()
    return items

or

class BaseResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)


class IngredientsResponse(BaseResponse):
    id: PydanticObjectId
    name: str
    used_in_drinks: List["IngredientsDrinksResponse"] = Field(default_factory=list)


class IngredientsDrinksResponse(BaseResponse):
    id: PydanticObjectId
    name: str

@router.get("/ingredients")
async def ingredients_items() -> list[IngredientsResponse]:
    items = await Ingredients.find_all(fetch_links=True, nesting_depth=1).to_list()
    return [IngredientsResponse.model_validate(item) for item in items]

[diegocgaona]

@hgalytoby thank you very much!

I wouldn't be able to solve it like you.

I still have one problem... how to do in the POST/PUT case?
The errors happens in the Ingredients(Document) which I use to create the item:

@router.post("/", response_model=IngredientsResponse)
async def create_item(
        item: Ingredients,  # This is the document, which causes the error even here
):
    created_item = await item.create() # If I try to use a modified BaseModel, it do not have .create()
    print(f"Created Item: {created_item}")
    return created_item

# Similar problem with PUT
@router.put("/{item_id}", response_model=IngredientsResponse)
async def update_item(
        item: Ingredients,
        item_id: PydanticObjectId,
        item_to_be_updated: Ingredients
):
    await item_to_be_updated.set(item.model_dump(exclude_unset=True))
    return await Ingredients.get(item_id, fetch_links=True, nesting_depth=1)

Thanks in advance

[diegocgaona]

I think I did it, don't know if is the best way...

# I created a Create BaseModel to use in Post without the BackLink
class IngredientsCreate(BaseModel):
    name: Indexed(str, unique=True)

# My POST uses it to the Json of the documentation and to validate, and then I convert it to the Document with BackLink and create the item
@router.post("/", response_model=IngredientsResponse)
async def create_item(
        item: IngredientsCreate,
):
    new_item_document = Ingredients(**item.model_dump())
    created_item = await new_item_document .create()

    return await Ingredients.get(created_item.id, fetch_links=True, nesting_depth=1, projection_model=IngredientsResponse)

Thanks!!

[hgalytoby]

There's nothing wrong with what you're doing. Other projects on the web are doing the same thing.

Create a CreateModel or UpdateModel and save the object as a Doc in the Database.

I'll give you an example, this is the SQL approach, you can see there are CreateModel and UpdateModel.

https://sqlmodel.tiangolo.com/tutorial/fastapi/update/#heroupdate-model

You can also look for MongoDB examples, or other SQL examples, that's how most people do it.