Webapp module

.github/workflows/test.yml

@@ -14,7 +14,7 @@ jobs:
     steps:
     - uses: actions/checkout@v4
     - name: Set up Python
-      uses: actions/setup-python@v2
+      uses: actions/setup-python@v5
       with:
         python-version:  ${{ matrix.python-version }}
     - name: Display Python version

.gitignore

@@ -1,3 +1,4 @@
+app.db
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

dapitains/app/app.py

@@ -0,0 +1,183 @@
+from typing import Dict, Any, Optional
+
+from sqlalchemy.orm.collections import collection
+
+from tests.test_db_create import test_navigation
+
+try:
+    import uritemplate
+    from flask import Flask, request, Response
+    from flask_sqlalchemy import SQLAlchemy
+    import click
+except ImportError:
+    print("This part of the package can only be imported with the web requirements.")
+    raise
+
+import json
+
+from dapitains.app.database import db, Collection, Navigation
+from dapitains.app.navigation import get_nav
+
+
+def msg_4xx(string, code=404) -> Response:
+    return Response(json.dumps({"message": string}), status=code, mimetype="application/json")
+
+
+def collection_view(
+        identifier: Optional[str],
+        nav: str,
+        templates: Dict[str, uritemplate.URITemplate]
+) -> Response:
+    """ Builds a collection view, regardless of how the parameters are received
+
+    :param identifier:
+    :param nav:
+    :param templates:
+    """
+    if not identifier:
+        coll: Collection = db.session.query(Collection).filter(~Collection.parents.any()).first()
+    else:
+        coll = Collection.query.where(Collection.identifier==identifier).first()
+    if coll is None:
+        return msg_4xx("Unknown collection")
+    out = coll.json()
+
+    if nav == 'children':
+        related_collections = db.session.query(Collection).filter(
+            Collection.parents.any(id=coll.id)
+        ).all()
+    elif nav == 'parents':
+        related_collections = db.session.query(Collection).filter(
+            Collection.children.any(id=coll.id)
+        ).all()
+    else:
+        return msg_4xx(f"nav parameter has a wrong value {nav}", code=400)
+
+    return Response(json.dumps({
+        "@context": "https://distributed-text-services.github.io/specifications/context/1-alpha1.json",
+        "dtsVersion": "1-alpha",
+        **out,
+        "totalParents": coll.total_parents,
+        "totalChildren": coll.total_children,
+        "collection": templates["collection"].uri,
+        "member": [
+                related.json(
+                    inject=dict(
+                        **{
+                            "collection": templates["collection"].partial({"id": related.identifier}).uri,
+                            "document": templates["document"].partial({"resource": related.identifier}).uri,
+                        },
+                        **(
+                            {
+                                "navigation": templates["navigation"].partial({"resource": related.identifier}).uri,
+                            } if coll.citeStructure else {}
+                        )
+                    ) if related.resource else related.json({
+                        "collection": templates["collection"].partial({"id": related.identifier}).uri
+                    })
+                )
+                for related in related_collections
+            ]
+    }), mimetype="application/json", status=200)
+
+
+def navigation_view(resource, ref, start, end, tree, down, templates: Dict[str, str]) -> Response:
+    if not resource:
+        return msg_4xx("Resource parameter was not provided")
+
+    collection: Collection = Collection.query.where(Collection.identifier == resource).first()
+    if not collection:
+        return msg_4xx(f"Unknown resource `{resource}`")
+
+    nav: Navigation = Navigation.query.where(Navigation.collection_id == collection.id).first()
+    if nav is None:
+        return msg_4xx(f"The resource `{resource}` does not support navigation")
+
+    # Check for forbidden combinations
+    if ref or start or end:
+        if tree not in nav.references:
+            return msg_4xx(f"Unknown tree {tree} for resource `{resource}`")
+        elif ref and (start or end):
+            return msg_4xx(f"You cannot provide a ref parameter as well as start or end", code=400)
+        elif not ref and ((start and not end) or (end and not start)):
+            return msg_4xx(f"Range is missing one of its parameters (start or end)", code=400)
+    else:
+        if down is None:
+            return msg_4xx(f"The down query parameter is required when requesting without ref or start/end", code=400)
+
+    refs = nav.references[tree]
+    paths = nav.paths[tree]
+    members, start, end = get_nav(refs=refs, paths=paths, start_or_ref=start or ref, end=end, down=down)
+    return Response(json.dumps({
+        "@context": "https://distributed-text-services.github.io/specifications/context/1-alpha1.json",
+        "dtsVersion": "1-alpha",
+        "@type": "Navigation",
+        "@id": "https://example.org/api/dts/navigation/?resource=https://en.wikisource.org/wiki/Dracula&down=1",
+        "resource": collection.json(inject=templates),  # To Do: implement and inject URI templates
+        "member": members
+    }), mimetype="application/json", status=200)
+
+
+def create_app(
+        app: Flask,
+        use_query: bool = False
+) -> (Flask, SQLAlchemy):
+    """
+
+    Initialisation of the DB is up to you
+    """
+    navigation_template = uritemplate.URITemplate("/navigation/{?resource}{&ref,start,end,tree,down}")
+    collection_template = uritemplate.URITemplate("/collection/collection/{?id,nav}")
+    document_template = uritemplate.URITemplate("/document/{?resource}{&ref,start,end,tree}")
+
+    @app.route("/collection/")
+    def collection_route():
+        resource = request.args.get("id")
+        nav = request.args.get("nav", "children")
+
+        return collection_view(resource, nav, templates={
+            "navigation": navigation_template,
+            "collection": collection_template,
+            "document": document_template,
+        })
+
+    @app.route("/navigation/")
+    def navigation_route():
+        resource = request.args.get("resource")
+        ref = request.args.get("ref")
+        start = request.args.get("start")
+        end = request.args.get("end")
+        tree = request.args.get("tree")
+        down = request.args.get("down", type=int, default=None)
+
+        return navigation_view(resource, ref, start, end, tree, down, templates={
+            "navigation": navigation_template.partial({"resource": resource}).uri,
+            "collection": collection_template.partial({"id": resource}).uri,
+            "document": document_template.partial({"resource": resource}).uri,
+        })
+
+    return app, db
+
+
+if __name__ == "__main__":
+    import os
+    from dapitains.app.ingest import store_catalog
+    from dapitains.metadata.xml_parser import parse
+
+    app = Flask(__name__)
+    _, db = create_app(app)
+
+    basedir = os.path.abspath(os.path.dirname(__file__))
+    db_path = os.path.join(basedir, 'app.db')
+    app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{db_path}'
+    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
+
+    db.init_app(app)
+    with app.app_context():
+        db.drop_all()
+        db.create_all()
+
+        catalog, _ = parse(f"{basedir}/../../tests/catalog/example-collection.xml")
+        store_catalog(catalog)
+
+    app.run()

dapitains/app/database.py

@@ -0,0 +1,151 @@
+from collections import defaultdict
+
+try:
+    from flask_sqlalchemy import SQLAlchemy
+    from sqlalchemy.ext.mutable import MutableDict, Mutable
+    from sqlalchemy.types import TypeDecorator, TEXT
+    from sqlalchemy import func
+    import click
+except ImportError:
+    print("This part of the package can only be imported with the web requirements.")
+    raise
+
+from typing import Optional, Dict, Any
+import dapitains.metadata.classes as abstracts
+import json
+
+
+class CustomKeyJSONDecoder(json.JSONDecoder):
+    def __init__(self, *args, **kwargs):
+        super().__init__(object_hook=self.object_hook, *args, **kwargs)
+
+    def object_hook(self, obj):
+        # Only convert 'None' string keys back to None
+        return {None if k == 'null' else k: v for k, v in obj.items()}
+
+db = SQLAlchemy()
+
+parent_child_association = db.Table('parent_child_association',
+    db.Column('parent_id', db.Integer, db.ForeignKey('collections.id'), primary_key=True),
+    db.Column('child_id', db.Integer, db.ForeignKey('collections.id'), primary_key=True)
+)
+
+
+class JSONEncoded(TypeDecorator):
+    """Enables JSON storage by encoding and decoding on the fly."""
+    impl = TEXT
+
+    def process_bind_param(self, value, dialect):
+        if value is None:
+            return None
+        else:
+            return json.dumps(value)
+
+    def process_result_value(self, value, dialect):
+        if value is None:
+            return None
+        return json.loads(value, cls=CustomKeyJSONDecoder)
+
+
+class Collection(db.Model):
+    __tablename__ = 'collections'
+
+    id = db.Column(db.Integer, primary_key=True, autoincrement=True, nullable=False)
+    identifier = db.Column(db.String, nullable=False, unique=True)
+    title = db.Column(db.String, nullable=False)
+    description = db.Column(db.String, nullable=True)
+    resource = db.Column(db.Boolean, default=False)
+    filepath = db.Column(db.String, nullable=True)
+    dublin_core = db.Column(JSONEncoded, nullable=True)
+    extensions = db.Column(JSONEncoded, nullable=True)
+    citeStructure = db.Column(JSONEncoded, nullable=True)
+    default_tree = db.Column(db.String, nullable=True)
+
+    # One-to-one relationship with Navigation
+    navigation = db.relationship('Navigation', uselist=False, backref='collection', lazy=True)
+
+    parents = db.relationship(
+        'Collection',
+        secondary=parent_child_association,
+        primaryjoin=id == parent_child_association.c.child_id,
+        secondaryjoin=id == parent_child_association.c.parent_id,
+        backref='children'
+    )
+
+    @property
+    def total_children(self):
+        return db.session.query(func.count(parent_child_association.c.child_id)).filter(
+            parent_child_association.c.parent_id == self.id
+        ).scalar()
+
+    @property
+    def total_parents(self):
+        return db.session.query(func.count(parent_child_association.c.parent_id)).filter(
+            parent_child_association.c.child_id == self.id
+        ).scalar()
+
+    def json(self, inject: Optional[Dict[str, Any]] = None):
+        data = {
+            "@type": "Resource" if self.resource else "Collection",
+            "@id": self.identifier,
+            "title": self.title,
+            **(inject or {})
+        }
+        if self.description:
+            data["description"] = self.description
+        if self.resource:
+            data["citationTrees"] = []
+        if self.citeStructure:
+            data["citationTrees"] = [self.citeStructure[self.default_tree]]
+            if len(self.citeStructure) >= 1:
+                data["citationTrees"][0]["identifier"] = self.default_tree
+                for key in self.citeStructure:
+                    if key != self.default_tree:
+                        data["citationTrees"].append(self.citeStructure[key])
+                        self.citeStructure[key]["identifier"] = key
+        if self.dublin_core:  # ToDo: Fix the way it's presented to adapt to dts view
+            data["dublinCore"] = self.dublin_core
+        if self.extensions:
+            data["extensions"] = self.extensions
+
+        return data
+
+    @classmethod
+    def from_class(cls, obj: abstracts.Collection) -> "Collection":
+        dublin_core = defaultdict(list)
+        for dublin in obj.dublin_core:
+            if dublin.language:
+                dublin_core[dublin.term].append({"lang": dublin.language, "value": dublin.value})
+            else:
+                dublin_core[dublin.term].append(dublin.value)
+
+        extensions = defaultdict(list)
+        for exte in obj.extensions:
+            if exte.language:
+                extensions[exte.term].append({"lang": exte.language, "value": exte.value})
+            else:
+                extensions[exte.term].append(exte.value)
+
+
+        obj = cls(
+            identifier=obj.identifier,
+            title=obj.title,
+            description=obj.description,
+            resource=obj.resource,
+            filepath=obj.filepath,
+            # We are dumping because it's not read or accessible
+            dublin_core=dublin_core,  #[dub.json() for dub in obj.dublin_core],
+            extensions=extensions,  # [ext.json() for ext in obj.extension]
+        )
+        return obj
+
+
+class Navigation(db.Model):
+    __tablename__ = 'navigations'
+
+    id = db.Column(db.Integer, primary_key=True, autoincrement=True, nullable=False)
+    collection_id = db.Column(db.Integer, db.ForeignKey('collections.id'), nullable=False, unique=True)
+
+    # JSON fields stored as TEXT
+    paths = db.Column(JSONEncoded, nullable=False, default={})
+    references = db.Column(JSONEncoded, nullable=False, default={})

dapitains/app/ingest.py

@@ -0,0 +1,46 @@
+from typing import Dict, Optional
+from dapitains.app.database import Collection, Navigation, db, parent_child_association
+from dapitains.app.navigation import generate_paths
+from dapitains.metadata.xml_parser import Catalog
+from dapitains.tei.document import Document
+import tqdm
+
+
+def store_single(catalog: Catalog, keys: Optional[Dict[str, int]]):
+    keys = keys or {}
+    for identifier, collection in tqdm.tqdm(catalog.objects.items(), desc="Parsing all collections"):
+        coll_db = Collection.from_class(collection)
+        db.session.add(coll_db)
+        db.session.flush()
+        keys[coll_db.identifier] = coll_db.id
+        if collection.resource:
+            doc = Document(collection.filepath)
+            if doc.citeStructure:
+                references = {
+                    tree: [ref.json() for ref in obj.find_refs(doc.xml, structure=obj.structure)]
+                    for tree, obj in doc.citeStructure.items()
+                }
+                paths = {key: generate_paths(tree) for key, tree in references.items()}
+                nav = Navigation(collection_id=coll_db.id, paths=paths, references=references)
+                db.session.add(nav)
+                coll_db.citeStructure = {
+                    key: value.structure.json()
+                    for key, value in doc.citeStructure.items()
+                }
+                coll_db.default_tree = doc.default_tree
+                db.session.add(coll_db)
+        db.session.commit()
+
+    for parent, child in catalog.relationships:
+        insert_statement = parent_child_association.insert().values(
+            parent_id=keys[parent],
+            child_id=keys[child]
+        )
+        db.session.execute(insert_statement)
+    db.session.commit()
+
+
+def store_catalog(*catalogs):
+    keys = {}
+    for catalog in catalogs:
+        store_single(catalog, keys)