feat: repository demo

.gitignore

@@ -0,0 +1,3 @@
+*.db
+.idea/
+Pipfile.lock

Pipfile

@@ -0,0 +1,13 @@
+[[source]]
+url = "https://pypi.org/simple"
+verify_ssl = true
+name = "pypi"
+
+[packages]
+sqlalchemy = "*"
+pydantic = "*"
+
+[dev-packages]
+
+[requires]
+python_version = "3.9"

app/db/base.py

@@ -0,0 +1,5 @@
+# Import all the models, so that Base has them before being
+# imported by Alembic
+from app.db.base_class import Base  # noqa
+from app.models.pipeline import Pipeline  # noqa
+from app.models.scenario import Scenario  # noqa

app/db/base_class.py

@@ -0,0 +1,13 @@
+from typing import Any
+
+from sqlalchemy.ext.declarative import as_declarative, declared_attr
+
+
+@as_declarative()
+class Base:
+    id: Any
+    __name__: str
+    # Generate __tablename__ automatically
+    @declared_attr
+    def __tablename__(cls) -> str:
+        return cls.__name__.lower()

app/db/init_db.py

@@ -0,0 +1,16 @@
+from sqlalchemy.orm import Session
+
+from app.db import base  # noqa: F401
+
+# make sure all SQL Alchemy models are imported (app.db.base) before initializing DB
+# otherwise, SQL Alchemy might fail to initialize relationships properly
+# for more details: https://github.com/tiangolo/full-stack-fastapi-postgresql/issues/28
+from app.db.base_class import Base
+from app.db.session import engine
+
+
+def init_db() -> None:
+    # Tables should be created with Alembic migrations
+    # But if you don't want to use migrations, create
+    # the tables un-commenting the next line
+    Base.metadata.create_all(bind=engine)

app/db/session.py

@@ -0,0 +1,6 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+
+engine = create_engine("sqlite:///test.db", pool_pre_ping=True)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

app/encoders/__init__.py

@@ -0,0 +1 @@
+from .encoders import jsonable_encoder

app/encoders/encoders.py

@@ -0,0 +1,170 @@
+from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
+from pydantic.json import ENCODERS_BY_TYPE
+from pydantic import BaseModel
+from collections import defaultdict
+
+
+import dataclasses
+from enum import Enum
+from pathlib import PurePath
+from types import GeneratorType
+
+SetIntStr = Set[Union[int, str]]
+DictIntStrAny = Dict[Union[int, str], Any]
+
+def generate_encoders_by_class_tuples(
+    type_encoder_map: Dict[Any, Callable[[Any], Any]]
+) -> Dict[Callable[[Any], Any], Tuple[Any, ...]]:
+    encoders_by_class_tuples: Dict[Callable[[Any], Any], Tuple[Any, ...]] = defaultdict(
+        tuple
+    )
+    for type_, encoder in type_encoder_map.items():
+        encoders_by_class_tuples[encoder] += (type_,)
+    return encoders_by_class_tuples
+
+
+encoders_by_class_tuples = generate_encoders_by_class_tuples(ENCODERS_BY_TYPE)
+
+def jsonable_encoder(
+    obj: Any,
+    include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
+    exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
+    by_alias: bool = True,
+    exclude_unset: bool = False,
+    exclude_defaults: bool = False,
+    exclude_none: bool = False,
+    custom_encoder: Optional[Dict[Any, Callable[[Any], Any]]] = None,
+    sqlalchemy_safe: bool = True,
+) -> Any:
+    custom_encoder = custom_encoder or {}
+    if custom_encoder:
+        if type(obj) in custom_encoder:
+            return custom_encoder[type(obj)](obj)
+        else:
+            for encoder_type, encoder_instance in custom_encoder.items():
+                if isinstance(obj, encoder_type):
+                    return encoder_instance(obj)
+    if include is not None and not isinstance(include, (set, dict)):
+        include = set(include)
+    if exclude is not None and not isinstance(exclude, (set, dict)):
+        exclude = set(exclude)
+    if isinstance(obj, BaseModel):
+        encoder = getattr(obj.__config__, "json_encoders", {})
+        if custom_encoder:
+            encoder.update(custom_encoder)
+        obj_dict = obj.dict(
+            include=include,  # type: ignore # in Pydantic
+            exclude=exclude,  # type: ignore # in Pydantic
+            by_alias=by_alias,
+            exclude_unset=exclude_unset,
+            exclude_none=exclude_none,
+            exclude_defaults=exclude_defaults,
+        )
+        if "__root__" in obj_dict:
+            obj_dict = obj_dict["__root__"]
+        return jsonable_encoder(
+            obj_dict,
+            exclude_none=exclude_none,
+            exclude_defaults=exclude_defaults,
+            custom_encoder=encoder,
+            sqlalchemy_safe=sqlalchemy_safe,
+        )
+    if dataclasses.is_dataclass(obj):
+        obj_dict = dataclasses.asdict(obj)
+        return jsonable_encoder(
+            obj_dict,
+            include=include,
+            exclude=exclude,
+            by_alias=by_alias,
+            exclude_unset=exclude_unset,
+            exclude_defaults=exclude_defaults,
+            exclude_none=exclude_none,
+            custom_encoder=custom_encoder,
+            sqlalchemy_safe=sqlalchemy_safe,
+        )
+    if isinstance(obj, Enum):
+        return obj.value
+    if isinstance(obj, PurePath):
+        return str(obj)
+    if isinstance(obj, (str, int, float, type(None))):
+        return obj
+    if isinstance(obj, dict):
+        encoded_dict = {}
+        allowed_keys = set(obj.keys())
+        if include is not None:
+            allowed_keys &= set(include)
+        if exclude is not None:
+            allowed_keys -= set(exclude)
+        for key, value in obj.items():
+            if (
+                (
+                    not sqlalchemy_safe
+                    or (not isinstance(key, str))
+                    or (not key.startswith("_sa"))
+                )
+                and (value is not None or not exclude_none)
+                and key in allowed_keys
+            ):
+                encoded_key = jsonable_encoder(
+                    key,
+                    by_alias=by_alias,
+                    exclude_unset=exclude_unset,
+                    exclude_none=exclude_none,
+                    custom_encoder=custom_encoder,
+                    sqlalchemy_safe=sqlalchemy_safe,
+                )
+                encoded_value = jsonable_encoder(
+                    value,
+                    by_alias=by_alias,
+                    exclude_unset=exclude_unset,
+                    exclude_none=exclude_none,
+                    custom_encoder=custom_encoder,
+                    sqlalchemy_safe=sqlalchemy_safe,
+                )
+                encoded_dict[encoded_key] = encoded_value
+        return encoded_dict
+    if isinstance(obj, (list, set, frozenset, GeneratorType, tuple)):
+        encoded_list = []
+        for item in obj:
+            encoded_list.append(
+                jsonable_encoder(
+                    item,
+                    include=include,
+                    exclude=exclude,
+                    by_alias=by_alias,
+                    exclude_unset=exclude_unset,
+                    exclude_defaults=exclude_defaults,
+                    exclude_none=exclude_none,
+                    custom_encoder=custom_encoder,
+                    sqlalchemy_safe=sqlalchemy_safe,
+                )
+            )
+        return encoded_list
+
+    if type(obj) in ENCODERS_BY_TYPE:
+        return ENCODERS_BY_TYPE[type(obj)](obj)
+    for encoder, classes_tuple in encoders_by_class_tuples.items():
+        if isinstance(obj, classes_tuple):
+            return encoder(obj)
+
+    try:
+        data = dict(obj)
+    except Exception as e:
+        errors: List[Exception] = []
+        errors.append(e)
+        try:
+            data = vars(obj)
+        except Exception as e:
+            errors.append(e)
+            raise ValueError(errors)
+    return jsonable_encoder(
+        data,
+        include=include,
+        exclude=exclude,
+        by_alias=by_alias,
+        exclude_unset=exclude_unset,
+        exclude_defaults=exclude_defaults,
+        exclude_none=exclude_none,
+        custom_encoder=custom_encoder,
+        sqlalchemy_safe=sqlalchemy_safe,
+    )

app/manager/pipeline_manager.py

@@ -0,0 +1,16 @@
+from app import repository
+from app.schemas.pipeline import Pipeline
+
+
+class PipelineManager:
+
+    def __init__(self):
+        self.repo = repository.pipeline_repo
+
+    def set(self, pipeline: Pipeline):
+        if _pipeline := self.repo.get_by_config(pipeline.config_id):
+            return _pipeline
+        self.repo.create(obj_in=pipeline)
+
+    def get(self, id: str):
+        return self.repo.get(id)

app/models/pipeline.py

@@ -0,0 +1,11 @@
+from sqlalchemy import Column, ForeignKey, Integer, String
+from sqlalchemy.orm import relationship
+
+from app.db.base_class import Base
+
+
+class Pipeline(Base):
+    id = Column(String, primary_key=True, index=True)
+    config_id = Column(String, index=True)
+    scenario_id = Column(Integer, ForeignKey("scenario.id"))
+    scenario = relationship("Scenario", back_populates="pipelines")

app/models/scenario.py

@@ -0,0 +1,12 @@
+from typing import TYPE_CHECKING
+
+from sqlalchemy import Column, ForeignKey, Integer, String
+from sqlalchemy.orm import relationship
+
+from app.db.base_class import Base
+
+
+class Scenario(Base):
+    id = Column(String, primary_key=True, index=True)
+    config_id = Column(String, index=True)
+    pipelines = relationship("Pipeline", back_populates="scenario")

app/repository/__init__.py

@@ -0,0 +1,5 @@
+from .repository_sql_pipeline import CRUDPipeline
+
+from app.models.pipeline import Pipeline
+
+pipeline_repo = CRUDPipeline(Pipeline)

app/repository/base_sql.py

@@ -0,0 +1,69 @@
+import uuid
+from typing import Any, Dict, Generic, List, Optional, Type, TypeVar, Union
+
+from app.encoders import jsonable_encoder
+from pydantic import BaseModel
+
+from app.db.base_class import Base
+from app.db.session import SessionLocal
+
+ModelType = TypeVar("ModelType", bound=Base)
+CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
+UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
+
+
+class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
+    def __init__(self, model: Type[ModelType]):
+        """
+        CRUD object with default methods to Create, Read, Update, Delete (CRUD).
+        **Parameters**
+        * `model`: A SQLAlchemy model class
+        * `schema`: A Pydantic model (schema) class
+        """
+        self.model = model
+        self.db = SessionLocal()
+
+    def get(self, id: Any) -> Optional[ModelType]:
+        return self.db.query(self.model).filter(self.model.id == id).first()
+
+    def get_by_config(self, config_id: Any) -> Optional[ModelType]:
+        return self.db.query(self.model).filter(self.model.config_id == config_id).first()
+
+    def get_multi(
+        self, *, skip: int = 0, limit: int = 100
+    ) -> List[ModelType]:
+        return self.db.query(self.model).offset(skip).limit(limit).all()
+
+    def create(self, *, obj_in: CreateSchemaType) -> ModelType:
+        obj_in_data = jsonable_encoder(obj_in)
+        obj_in_data["id"] = str(uuid.uuid4())
+        db_obj = self.model(**obj_in_data)  # type: ignore
+        self.db.add(db_obj)
+        self.db.commit()
+        self.db.refresh(db_obj)
+        return db_obj
+
+    def update(
+        self,
+        *,
+        db_obj: ModelType,
+        obj_in: Union[UpdateSchemaType, Dict[str, Any]]
+    ) -> ModelType:
+        obj_data = jsonable_encoder(db_obj)
+        if isinstance(obj_in, dict):
+            update_data = obj_in
+        else:
+            update_data = obj_in.dict(exclude_unset=True)
+        for field in obj_data:
+            if field in update_data:
+                setattr(db_obj, field, update_data[field])
+        self.db.add(db_obj)
+        self.db.commit()
+        self.db.refresh(db_obj)
+        return db_obj
+
+    def remove(self, *, id: int) -> ModelType:
+        obj = self.db.query(self.model).get(id)
+        self.db.delete(obj)
+        self.db.commit()
+        return obj

app/repository/repository_sql_pipeline.py

@@ -0,0 +1,8 @@
+from app.repository.base_sql import CRUDBase
+from app.models.pipeline import Pipeline
+from app.schemas.pipeline import PipelineCreate, PipelineUpdate
+
+
+class CRUDPipeline(CRUDBase[Pipeline, PipelineCreate, PipelineUpdate]):
+    pass
+

app/repository/repository_sql_scenario.py

@@ -0,0 +1,13 @@
+from typing import List
+
+
+from app.repository.base_sql import CRUDBase
+from app.models.scenario import Scenario
+from app.schemas.scenario import ScenarioCreate, ScenarioUpdate
+
+
+class CRUDScenario(CRUDBase[Scenario, ScenarioCreate, ScenarioUpdate]):
+    pass
+
+
+scenario = CRUDScenario(Scenario)