Source code for brickschema.orm

"""
ORM for Brick
"""
from . import namespaces as ns

try:
    from sqlalchemy import Column, String, ForeignKey, create_engine
    from sqlalchemy.orm import relationship, sessionmaker
    from sqlalchemy.ext.declarative import declarative_base
except ImportError:
    print(
        "SQLAlchemy not found. Please install brickschema with the 'orm' option:\n\n\tpip install brickschema[orm]"
    )
    import sys

    sys.exit(1)

Base = declarative_base()
# TODO: brick:feeds (many-to-many), brick:hasPart


[docs]class Equipment(Base): """ SQLAlchemy ORM class for BRICK.Equipment; see SQLORM class for usage """ __tablename__ = "equipment" name = Column(String, primary_key=True) type = Column(String) points = relationship("Point", back_populates="equipment") location_id = Column(ForeignKey("location.name")) location = relationship("Location", back_populates="equipment")
[docs]class Point(Base): """ SQLAlchemy ORM class for BRICK.Point; see SQLORM class for usage """ __tablename__ = "point" name = Column(String, primary_key=True) type = Column(String) equipment_id = Column(ForeignKey("equipment.name")) equipment = relationship("Equipment", back_populates="points") location_id = Column(ForeignKey("location.name")) location = relationship("Location", back_populates="points")
[docs]class Location(Base): """ SQLAlchemy ORM class for BRICK.Location; see SQLORM class for usage """ __tablename__ = "location" name = Column(String, primary_key=True) type = Column(String) equipment = relationship("Equipment", back_populates="location") points = relationship("Point", back_populates="location")
[docs]class SQLORM: """ A SQLAlchemy-based ORM for Brick models. Currently, the ORM models Locations, Points and Equipment and the basic relationships between them. """ def __init__(self, graph, connection_string="sqlite://brick_orm.db"): """ Creates a new ORM instance over the given Graph using SQLAlchemy. The ORM does not capture *all* information expressed in a Brick model, but can be easily extended over time to capture more information. Args: graph (brickschema.Graph): a Brick schema graph containing instances we want to interact with. **Note**: this graph should not have any inference applied to it (RDFS or otherwise) connection_string (str): a database URL telling SQLAlchemy how to connect to the database that is backing the ORM. See [SQLAlchemy's documentation on database URLs](https://docs.sqlalchemy.org/en/13/core/engines.html#database-urls) """ self._graph = graph self._engine = create_engine(connection_string) Base.metadata.create_all(self._engine) # the SQLAlchemy session; use for queries, etc self.session = sessionmaker(bind=self._engine)() # populate the database # get all equipment res = self._graph.query( """SELECT ?equip ?type WHERE { ?equip rdf:type/rdfs:subClassOf* brick:Equipment . ?equip rdf:type ?type }""" ) for (equip_name, equip_type) in res: equip = Equipment(name=equip_name, type=equip_type) self.session.merge(equip) # get all points of equipment res = self._graph.query( """SELECT ?point ?type ?equip WHERE { ?point rdf:type/rdfs:subClassOf* brick:Point . ?point rdf:type ?type . { ?point brick:isPointOf ?equip . } UNION { ?equip brick:hasPoint ?point . } }""" ) for (point_name, point_type, equip_name) in res: point = Point(name=point_name, type=point_type, equipment_id=equip_name) self.session.merge(point) # get all locations res = self._graph.query( """SELECT ?location ?type WHERE { ?location rdf:type/rdfs:subClassOf* brick:Location . ?location rdf:type ?type . }""" ) for (loc_name, loc_type) in res: loc = Location(name=loc_name, type=loc_type) self.session.merge(loc) # get all locations of equipment res = self._graph.query( """SELECT ?location ?type ?equip WHERE { ?equip rdf:type/rdfs:subClassOf* brick:Equipment . ?location rdf:type/rdfs:subClassOf* brick:Location . ?location rdf:type ?type . { ?location brick:isLocationOf ?equip . } UNION { ?equip brick:hasLocation ?location . } }""" ) for (loc_name, loc_type, equip_name) in res: # get existing equip object equip = ( self.session.query(Equipment).filter(Equipment.name == equip_name).one() ) # get existing Location loc = self.session.query(Location).filter(Location.name == loc_name).one() self.session.merge(loc) loc.equipment.append(equip) self.session.merge(loc) self.session.commit()
class _DynamicORM: """ TODO: under construction Object-oriented interface to Brick models. We turn triples into objects as follows: 1. Equipment with points: all instances of equipment """ def __init__(self, graph): """ Creates a new ORM instance over the given Graph Args: graph (brickschema.Graph): a Brick schema graph containing instances we want to interact with. **Note**: this graph should not have any inference applied to it (RDFS or otherwise) """ self.graph = graph # construct initial Equipment class self.equipment_classes = { "Equipment": type( "Equipment", (), { "URI": ns.BRICK["Equipment"], "classname": "Equipment", "name": None, "__repr__": _brick_repr, "points": [], }, ), } # construct subclasses recursively self._build_subclasses( self.equipment_classes["Equipment"], self.equipment_classes, set() ) self.point_classes = { "Point": type( "Point", (), {"URI": ns.BRICK["Point"], "classname": "Point", "name": None}, ), } self._build_subclasses(self.point_classes["Point"], self.point_classes, set()) # get equipment instances self.instances = {} for name, klass in self.equipment_classes.items(): res = self.graph.query( f"""SELECT ?inst ?point ?pointtype WHERE {{ ?inst a <{klass.URI}> . ?inst brick:hasPoint ?point . ?point a ?pointtype }}""" ) for (inst, point, pointtype) in res: inst_name = inst.split("#")[-1] class_inst = self.instances.get(inst_name, klass()) pointtype = pointtype.split("#")[-1] if pointtype not in self.point_classes: print(f"No instances of {pointtype} for {klass}") continue point_class = self.point_classes[pointtype] point_inst = point_class() point_inst.name = point class_inst.name = inst_name class_inst.points.append(point_inst) self.instances[inst_name] = class_inst def _build_subclasses(self, rootclass, dest, visited=None): if rootclass.URI in visited: return visited.add(rootclass.URI) res = self.graph.query( f"""SELECT ?class WHERE {{ ?class rdfs:subClassOf <{rootclass.URI}> }}""" ) for row in res: class_uri = row[0] name = class_uri.split("#")[-1] klass = type( name, (rootclass,), { "URI": class_uri, "classname": name, "__repr__": _brick_repr, "name": None, "points": [], }, ) dest[name] = klass self._build_subclasses(klass, dest, visited=visited) def _brick_repr(self): return f"<BRICK {self.classname}: {self.name}>"