database

Author: MaxNumerique

build/lib/opengeodeweb_microservice/__init__.py

@@ -0,0 +1,32 @@
+"""Database connection management"""
+
+from typing import Optional
+from sqlalchemy.orm import scoped_session
+from flask import Flask
+from flask_sqlalchemy import SQLAlchemy
+from flask_sqlalchemy.session import Session
+from ..microservice.base import Base
+
+DATABASE_FILENAME = "project.db"
+db: Optional[SQLAlchemy] = None
+
+
+def init_database(app: Flask, db_filename: str = DATABASE_FILENAME) -> SQLAlchemy:
+    global db
+    if db is None:
+        db = SQLAlchemy(model_class=Base)
+    db.init_app(app)
+    with app.app_context():
+        db.create_all()
+    return db
+
+
+def get_database() -> Optional[SQLAlchemy]:
+    return db
+
+
+get_database_connection = get_database
+
+
+def get_session() -> Optional[scoped_session[Session]]:
+    return db.session if db else None

build/lib/opengeodeweb_microservice/database/__init__.py

@@ -0,0 +1,7 @@
+from flask import Flask
+from flask_sqlalchemy import SQLAlchemy
+from sqlalchemy.orm import DeclarativeBase
+
+
+class Base(DeclarativeBase):
+    pass

build/lib/opengeodeweb_microservice/database/connection.py

@@ -0,0 +1,52 @@
+from sqlalchemy import String, JSON
+from sqlalchemy.orm import Mapped, mapped_column
+from ..database.connection import get_session
+from .base import Base
+import uuid
+
+
+class Data(Base):
+    __tablename__ = "datas"
+
+    id: Mapped[str] = mapped_column(
+        String, primary_key=True, default=lambda: str(uuid.uuid4()).replace("-", "")
+    )
+    native_file_name: Mapped[str] = mapped_column(String, nullable=False)
+    viewable_file_name: Mapped[str] = mapped_column(String, nullable=False)
+    geode_object: Mapped[str] = mapped_column(String, nullable=False)
+
+    light_viewable: Mapped[str | None] = mapped_column(String, nullable=True)
+    input_file: Mapped[str | None] = mapped_column(String, nullable=True)
+    additional_files: Mapped[list[str] | None] = mapped_column(JSON, nullable=True)
+
+    @staticmethod
+    def create(
+        geode_object: str,
+        input_file: str | None = None,
+        additional_files: list[str] | None = None,
+    ) -> "Data":
+        input_file = input_file if input_file is not None else ""
+        additional_files = additional_files if additional_files is not None else []
+
+        data_entry = Data(
+            geode_object=geode_object,
+            input_file=input_file,
+            additional_files=additional_files,
+            native_file_name="",
+            viewable_file_name="",
+            light_viewable=None,
+        )
+
+        session = get_session()
+        if session:
+            session.add(data_entry)
+            session.flush()
+            session.commit()
+        return data_entry
+
+    @staticmethod
+    def get(data_id: str) -> "Data | None":
+        session = get_session()
+        if session:
+            return session.get(Data, data_id)
+        return None

build/lib/opengeodeweb_microservice/microservice/__init__.py

@@ -1,8 +1,8 @@
 import os
 import pytest
 from flask import Flask
-from src.opengeodeweb_microservice.database.connection import init_database, get_session
-from src.opengeodeweb_microservice.microservice.data import Data
+from opengeodeweb_microservice.database.connection import init_database, get_session
+from opengeodeweb_microservice.microservice.data import Data
 
 
 @pytest.fixture(scope="session")