Database

Database adapters for curies.

Using `sqlmodel`

SQLModel is a joint abstraction over pydantic and sqlalchemy. If you want to use curies.Reference within a SQLModel class, you can do so by setting the SQLAlchemy column type through the sa_column keyword argument to the results returned by get_reference_sa_column(),

For example, this can be used to model a semantic triple, which has a subject reference, predicate reference, and object reference like in the following:

from curies import Reference
from curies.database import get_reference_sa_column
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Edge(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    subject: Reference = Field(sa_column=get_reference_sa_column())
    predicate: Reference = Field(sa_column=get_reference_sa_column())
    object: Reference = Field(sa_column=get_reference_sa_column())


e1 = Edge(subject="CHEBI:135122", predicate="skos:exactMatch", object="mesh:C073738")
e2 = Edge(subject="CHEBI:135125", predicate="skos:exactMatch", object="mesh:C073260")

engine = create_engine("sqlite://")
SQLModel.metadata.create_all(engine)

# Add edges to the database
with Session(engine) as session:
    session.add_all([e1, e2])
    session.commit()

# Query for edges with a given subject, by string
with Session(engine) as session:
    statement = select(Edge).where(Edge.subject == "CHEBI:135122")
    edges = session.exec(statement).all()

# Query for edges with a given subject, by string
with Session(engine) as session:
    statement = select(Edge).where(
        Edge.subject == Reference(prefix="CHEBI", identifier="135125")
    )
    edges = session.exec(statement).all()

If you want to have a field that has an optional reference, you can do it as expected using the same column type, such as for the author field below:

class Edge(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    subject: Reference = Field(sa_column=get_reference_sa_column())
    predicate: Reference = Field(sa_column=get_reference_sa_column())
    object: Reference = Field(sa_column=get_reference_sa_column())
    author: Reference | None = Field(sa_column=get_reference_sa_column())

If you want to have a field with a list of references, you can do it using get_reference_list_sa_column():

class Record(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    authors: list[Reference] = Field(
        default_factory=list, sa_column=get_reference_list_sa_column()
    )

This also works how you would expect if you don’t use a default_factory or if you choose to make it optional list[Reference] | None.

Using `sqlalchemy`

SQLAlchemy is a combine high- and mid-level database abstraction layer and object-relational mapping. It has more opportunities for configuration over SQLModel.

The following example re-formulates the above example for SQLModel using the slightly more direct SQLAlchemy formulation:

from curies import Reference
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import DeclarativeBase, Session


class Base(DeclarativeBase):
    pass


class Edge(Base):
    __tablename__ = "edge"

    id = Column(Integer, primary_key=True)

    subject = get_reference_sa_column()
    predicate = get_reference_sa_column()
    object = get_reference_sa_column()


e1 = Edge(subject="CHEBI:135122", predicate="skos:exactMatch", object="mesh:C073738")
e2 = Edge(subject="CHEBI:135125", predicate="skos:exactMatch", object="mesh:C073260")

engine = create_engine("sqlite://")
Base.metadata.create_all(engine)

# Add edges to the database
with Session(engine) as session:
    session.add_all([e1, e2])
    session.commit()

# Query for edges with a given subject, by string
with Session(engine) as session:
    statement = select(Edge).where(Edge.subject == "CHEBI:135122")
    edges = session.exec(statement).all()

# Query for edges with a given subject, by string
with Session(engine) as session:
    statement = select(Edge).where(
        Edge.subject == Reference(prefix="CHEBI", identifier="135125")
    )
    edges = session.exec(statement).all()

In the following example, SQLAlchemy is used with a composite column, where the prefix and identifier are given their own column, then a sqlalchemy.Composite object is used to tell the ORM to map both the prefix and identifier together into a single column, that exposes an appropriate curies.Reference class.

from curies import Reference
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import DeclarativeBase, Session


class Base(DeclarativeBase):
    pass


class Edge(Base):
    __tablename__ = "edge"

    id = Column(Integer, primary_key=True)

    subject_prefix = Column(String, nullable=False)
    subject_identifier = Column(String, nullable=False)
    predicate_prefix = Column(String, nullable=False)
    predicate_identifier = Column(String, nullable=False)
    object_prefix = Column(String, nullable=False)
    object_identifier = Column(String, nullable=False)

    subject = get_reference_sa_composite(subject_prefix, subject_identifier)
    predicate = get_reference_sa_composite(predicate_prefix, predicate_identifier)
    object = get_reference_sa_composite(object_prefix, object_identifier)


e1 = Edge(subject="CHEBI:135122", predicate="skos:exactMatch", object="mesh:C073738")
e2 = Edge(subject="CHEBI:135125", predicate="skos:exactMatch", object="mesh:C073260")

engine = create_engine("sqlite://")
Base.metadata.create_all(engine)

# Add edges to the database
with Session(engine) as session:
    session.add_all([e1, e2])
    session.commit()

# Query for edges with a given subject, by string
with Session(engine) as session:
    statement = select(Edge).where(Edge.subject == "CHEBI:135122")
    edges = session.exec(statement).all()

# Query for edges with a given subject, by string
with Session(engine) as session:
    statement = select(Edge).where(
        Edge.subject == Reference(prefix="CHEBI", identifier="135125")
    )
    edges = session.exec(statement).all()

Similarly to above, lists can be constructed using get_reference_list_sa_column():

class Record(Base):
    __tablename__ = "record"

    id = Column(Integer, primary_key=True)

    authors = get_reference_list_sa_column()

Functions

`get_reference_list_sa_column`(args, *kwargs)	Get a SQLAlchemy column with the type decorator for a :list of mod:curies.Reference.
`get_reference_sa_column`(args, *kwargs)	Get a SQLAlchemy column with the type decorator for a `curies.Reference`.
`get_reference_sa_composite`(prefix_column, ...)	Get a composite for a reference.

Classes

`SAReferenceListTypeDecorator`(args, *kwargs)	A SQLAlchemy type decorator for a list of `curies.Reference`.
`SAReferenceTypeDecorator`(args, *kwargs)	A SQLAlchemy type decorator for a `curies.Reference`.

Database

Using sqlmodel

Using sqlalchemy

Functions

Classes

Using `sqlmodel`

Using `sqlalchemy`