Map attributes from link model(association table)

[88Ocelot]

Example Code

class Employee(SQLModel, table=True):
    id: uuid.UUID | None = Field(default=uuid.uuid4, primary_key=True)
    username: str
    first_name: str
    last_name: str
    email: str = Field(regex=r"^[a-zA-Z0-9_.]+@\w+\.\w+$")
    is_active: bool

    departments: List["Department"] = Relationship(
        back_populates="employees", link_model=DepartmentEmployeeLink
    )

class DepartmentEmployeeLink(SQLModel, table=True):
    department_id: int | None = Field(
        default=None, foreign_key="department.id", primary_key=True
    )
    employee_id: uuid.UUID | None = Field(
        default=None, foreign_key="employee.id", primary_key=True
    )
    primary: bool


class Department(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    employees: List["Employee"] = Relationship(
        back_populates="departments", link_model=DepartmentEmployeeLink
    )

Description

How i can map primary attribute to employees departments list?

Operating System

Linux

Operating System Details

No response

SQLModel Version

0.0.6

Python Version

3.10

Additional Context

No response

[wscullen]

I am also struggling with this. I want to include the link model additional properties on the response object, but there is no clear example of how to modify a response model with additional fields from child model.

Edit:

Just was able to figure it out, this is how I did it using from_orm to generate a copy of the db model with the child models that have the additional properties that are stored in the link model.

def get_roi_by_id(roi_id: str, session: Session):
    logger.info(f"Getting all RoI by id {roi_id}")

    roi = session.exec(
        select(RegionOfInterest).where(RegionOfInterest.id == roi_id)
    ).one()

    roi_links = session.exec(
        select(TileRoiLink).where(roi.id == TileRoiLink.roi_id)
    ).all()

    new_roi = RegionOfInterestRead.from_orm(roi)
    logger.info(new_roi)

    # populate selected and highlighted based on TileRoiLink table
    for idx, tile in enumerate(new_roi.tiles):
        logger.info(idx)
        for roi_link in roi_links:
            if roi_link.dict()["tile_id"] == tile.id:
                logger.info("found matching tile link")
                setattr(tile, "selected", roi_link.selected)
                setattr(tile, "highlighted", roi_link.highlighted)

    return new_roi

[northtree]

The following example is from https://sqlmodel.tiangolo.com/tutorial/many-to-many/link-with-extra-fields/

from typing import List, Optional

from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select


class HeroTeamLink(SQLModel, table=True):
    team_id: Optional[int] = Field(
        default=None, foreign_key="team.id", primary_key=True
    )
    hero_id: Optional[int] = Field(
        default=None, foreign_key="hero.id", primary_key=True
    )
    is_training: bool = False

    team: "Team" = Relationship(back_populates="hero_links")
    hero: "Hero" = Relationship(back_populates="team_links")


class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str

    hero_links: List[HeroTeamLink] = Relationship(back_populates="team")


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

    team_links: List[HeroTeamLink] = Relationship(back_populates="hero")


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url, echo=True)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def create_heroes():
    with Session(engine) as session:
        team_preventers = Team(name="Preventers", headquarters="Sharp Tower")
        team_z_force = Team(name="Z-Force", headquarters="Sister Margaret’s Bar")

        hero_deadpond = Hero(
            name="Deadpond",
            secret_name="Dive Wilson",
        )
        hero_rusty_man = Hero(
            name="Rusty-Man",
            secret_name="Tommy Sharp",
            age=48,
        )
        hero_spider_boy = Hero(
            name="Spider-Boy",
            secret_name="Pedro Parqueador",
        )
        deadpond_team_z_link = HeroTeamLink(team=team_z_force, hero=hero_deadpond)
        deadpond_preventers_link = HeroTeamLink(
            team=team_preventers, hero=hero_deadpond, is_training=True
        )
        spider_boy_preventers_link = HeroTeamLink(
            team=team_preventers, hero=hero_spider_boy, is_training=True
        )
        rusty_man_preventers_link = HeroTeamLink(
            team=team_preventers, hero=hero_rusty_man
        )

        session.add(deadpond_team_z_link)
        session.add(deadpond_preventers_link)
        session.add(spider_boy_preventers_link)
        session.add(rusty_man_preventers_link)
        session.commit()

        for link in team_z_force.hero_links:
            print("Z-Force hero:", link.hero, "is training:", link.is_training)

        for link in team_preventers.hero_links:
            print("Preventers hero:", link.hero, "is training:", link.is_training)


def update_heroes():
    with Session(engine) as session:
        hero_spider_boy = session.exec(
            select(Hero).where(Hero.name == "Spider-Boy")
        ).one()
        team_z_force = session.exec(select(Team).where(Team.name == "Z-Force")).one()

        spider_boy_z_force_link = HeroTeamLink(
            team=team_z_force, hero=hero_spider_boy, is_training=True
        )
        team_z_force.hero_links.append(spider_boy_z_force_link)
        session.add(team_z_force)
        session.commit()

        print("Updated Spider-Boy's Teams:", hero_spider_boy.team_links)
        print("Z-Force heroes:", team_z_force.hero_links)

        for link in hero_spider_boy.team_links:
            if link.team.name == "Preventers":
                link.is_training = False

        session.add(hero_spider_boy)
        session.commit()

        for link in hero_spider_boy.team_links:
            print("Spider-Boy team:", link.team, "is training:", link.is_training)


def main():
    create_db_and_tables()
    create_heroes()
    update_heroes()


if __name__ == "__main__":
    main()

[wscullen]

@northtree I've read the documentation for that section, the key point I needed was this, which wasn't in that section:

 new_roi = RegionOfInterestRead.from_orm(roi)

I needed to create a non-table version of my model, one that had the two properties stored in the link model, selected and highlighted. The section you copy and pasted doesn't have an example for fetching data, only for creating and updating. For my question, I was trying to figure out how to return the link model only properties as though they were actually properties of one of the models being linked together. Maybe I missed it but I don't see that being explained in what you posted.

[northtree]

The example code should resolve @88Ocelot 's question.

@wscullen could you share your model definition

[wscullen]

I'm still struggling with this. @northtree I would really appreciate your help on this.

Here's my models:

class TileRoiLink(CamelModel, table=True):
    tile_id: Optional[str] = Field(
        default=None,
        foreign_key="tile.id",
        primary_key=True,
        nullable=False,
        index=True,
    )
    roi_id: Optional[str] = Field(
        default=None,
        foreign_key="regionofinterest.id",
        primary_key=True,
        nullable=False,
        index=True,
    )
    selected: bool = False # Extra properties that I want in the final query object returned to the user (Roi.tiles[0].selected)
    highlighted: bool = False # Extra properties that I want in the final query object returned to the user (Roi.tiles[0].highlighted)


class TileCollectionLink(CamelModel, table=True):
    tile_id: Optional[str] = Field(
        default=None, foreign_key="tile.id", primary_key=True, nullable=False
    )
    collection_id: Optional[str] = Field(
        default=None,
        foreign_key="collection.id",
        primary_key=True,
        nullable=False,
    )


# Tile models

class TileBase(CamelModel):
    vendor_id: str
    start_date: datetime
    end_date: datetime
    bbox: str
    footprint: str
    provider: str
    instrument: str
    platform: str = Field(default=None, nullable=True)
    properties: dict = Field(sa_column=Column(JSON, default={}))
    links: Optional[List[str]] = Field(sa_column=Column(ARRAY(String)))
    assets: Optional[List[str]] = Field(sa_column=Column(ARRAY(String)))
    thumbnail_url: str
    thumbnail_large_url: Optional[str] = None
    s3_url: Optional[str] = None
    downloaded: bool = False


class Tile(TileBase, table=True):
    id: Optional[str] = Field(default=None, primary_key=True, nullable=False)
    collections: List["Collection"] = Relationship(
        back_populates="tiles", link_model=TileCollectionLink
    )
    rois: List["RegionOfInterest"] = Relationship(
        back_populates="tiles", link_model=TileRoiLink
    )

class TileRead(TileBase):
    id: str
    collections: Optional[List["CollectionRead"]] = None
    job: Optional["JobRead"] = None
    selected: bool = False
    highlighted: bool = False

class RegionOfInterestBase(CamelModel):
    title: str = Field(sa_column=Column("title", String, unique=True))
    description: Optional[str] = None
    start_date: datetime
    end_date: datetime
    bbox: str
    footprint: dict = Field(sa_column=Column(JSON, default={}))
    updated_on: Optional[datetime] = None
    shared: bool
    owner_id: str = Field(default=None, foreign_key="user.id")
    created_on: datetime = Field(
        sa_column=Column(
            "created_on", DateTime(timezone=True), server_default=func.now()
        )
    )
    wrs_overlay: dict = Field(sa_column=Column(JSON, default={}))
    mgrs_overlay: dict = Field(sa_column=Column(JSON, default={}))

class RegionOfInterest(RegionOfInterestBase, table=True):
    id: Optional[str] = Field(default=None, primary_key=True, nullable=False)

    owner: User = Relationship(back_populates="rois")

    tiles: List["Tile"] = Relationship(back_populates="rois", link_model=TileRoiLink)
    jobs: List["Job"] = Relationship(back_populates="roi")
    layers: List["Layer"] = Relationship(back_populates="roi")

class RegionOfInterestReadSummary(RegionOfInterestBase):
    id: str

class RegionOfInterestRead(RegionOfInterestBase):
    id: str
    owner: UserRead
    tiles: List["TileRead"]
    jobs: Optional[List["JobRead"]] = None
    layers: Optional[List["LayerRead"]] = None

And the current solution for populating the "selected" and "highlighted" columns for each "tile" in the "tiles" list in the RegionOfInterest object being queried:

def get_roi_by_id(roi_id: str, session: Session):
    logger.info(f"Getting all RoI by id {roi_id}")

    roi = session.exec(
        select(RegionOfInterest).where(RegionOfInterest.id == roi_id)
    ).one()

    roi_links = session.exec(
        select(TileRoiLink).where(roi.id == TileRoiLink.roi_id)
    ).all()

    new_roi = RegionOfInterestRead.from_orm(roi)

    # populate selected and highlighted based on RoITileLink table
    for idx, tile in enumerate(new_roi.tiles):
        for roi_link in roi_links:
            if roi_link.dict()["tile_id"] == tile.id:
                setattr(tile, "selected", roi_link.selected)
                setattr(tile, "highlighted", roi_link.highlighted)

    return new_roi

Basically, is there a better way to populate the selected and highlighted columns of each tile for the selected Region of interest? Like a join on the TileRoiLink table with the tiles for the RoI?

I'm looking at other ways of loading Tiles separate from a relationship that links tiles to RegionsOfInterest, like returning a calculated column of dates that can be used to query the tiles directly for a region of interest, instead of returning all tiles up front.

Just looking for the correct most performant way to populate the selected and highlighted columns on each tile in the tiles list on the queried reqion of interest object. Thanks for taking the time to look at this.

[fmillama]

@wscullen did you manage to solve this in the end? I'm facing the same problem