Implemented entities_metadata_statements() for SQLImplementation()

src/oaklib/implementations/sqldb/sql_implementation.py

@@ -127,6 +127,7 @@
     DEFINITION,
     LANGUAGE_TAG,
     METADATA_MAP,
+    METADATA_STATEMENT,
     PRED_CURIE,
     PREFIX_MAP,
     RELATIONSHIP,
@@ -671,6 +672,29 @@ def entity_metadata_map(self, curie: CURIE, include_all_triples=False) -> METADA
         self.add_missing_property_values(curie, m)
         return dict(m)
 
+    def entities_metadata_statements(
+        self, curies: Iterable[CURIE], predicates: Optional[List[PRED_CURIE]] = None
+    ) -> Iterator[METADATA_STATEMENT]:
+        q = self.session.query(Statements)
+        # if not include_all_triples:
+        #     subquery = self.session.query(RdfTypeStatement.subject).filter(
+        #         RdfTypeStatement.object == "owl:AnnotationProperty"
+        #     )
+        #     annotation_properties = {row.subject for row in subquery}
+        #     annotation_properties = annotation_properties.union(STANDARD_ANNOTATION_PROPERTIES)
+        #     q = q.filter(Statements.predicate.in_(tuple(annotation_properties)))
+        q = q.filter(Statements.subject.in_(curies))
+        if predicates is not None:
+            q = q.filter(Statements.predicate.in_(predicates))
+        for row in q:
+            if row.value is not None:
+                v = _python_value(row.value, row.datatype)
+            elif row.object is not None:
+                v = row.object
+            else:
+                v = None
+            yield row.subject, row.predicate, v, row.datatype, {}
+
     def ontologies(self) -> Iterable[CURIE]:
         for row in self.session.query(OntologyNode):
             yield row.id

tests/test_implementations/test_sqldb.py

@@ -19,6 +19,7 @@
     HAS_PART,
     IS_A,
     LABEL_PREDICATE,
+    OIO_CREATION_DATE,
     PART_OF,
     RDF_TYPE,
 )
@@ -39,6 +40,7 @@
     HUMAN,
     IMBO,
     INPUT_DIR,
+    MEMBRANE,
     NUCLEAR_ENVELOPE,
     NUCLEUS,
     OUTPUT_DIR,
@@ -899,3 +901,11 @@ def test_transitive_object_properties(self):
 
     def test_simple_subproperty_of_chains(self):
         self.compliance_tester.test_simple_subproperty_of_chains(self.oi)
+
+    def test_entities_metadata_statements(self):
+        curies = [MEMBRANE]
+        predicates = [OIO_CREATION_DATE]
+        oi = self.oi
+        result = list(oi.entities_metadata_statements(curies=curies, predicates=predicates))
+        self.assertEqual(len(result), 1)
+        self.assertEqual(len(result[0]), 5)