Connected service management front to back; migrated to formik

api/schema/definitions/service.py

@@ -0,0 +1,105 @@
+import strawberry
+import typing
+import api.models as models
+from sqlalchemy.orm import Session
+
+if typing.TYPE_CHECKING:
+    from strawberry.types import Info
+    from api.schema import Query
+
+
+@strawberry.input
+class AddLoadbalancerInput:
+    servers: typing.List[str]
+
+
+@strawberry.input
+class AddServiceInput:
+    name: str
+    loadbalancer: AddLoadbalancerInput
+
+
+@strawberry.input
+class ServiceFilterInput:
+    id: typing.Optional[int] = None
+    search: typing.Optional[str] = None
+
+
+@strawberry.type
+class LoadbalancerServer:
+    id: int
+    address: str
+
+    @classmethod
+    def from_instance(cls, instance: models.LoadbalancerServers):
+        return cls(id=instance.id, address=instance.address)
+
+
+@strawberry.type
+class Loadbalancer:
+    servers: typing.List[LoadbalancerServer]
+
+    @classmethod
+    def from_instance(cls, instance: models.Loadbalancer):
+        if not instance:
+            return cls(servers=[])
+
+        return cls(
+            servers=[
+                LoadbalancerServer.from_instance(server) for server in instance.servers
+            ]
+        )
+
+
+@strawberry.type
+class Service:
+    id: int
+    name: str
+    loadbalancer: Loadbalancer
+
+    @classmethod
+    def from_instance(cls, instance: models.Service):
+        return cls(
+            id=instance.id,
+            name=instance.name,
+            loadbalancer=Loadbalancer.from_instance(instance.loadbalancer),
+        )
+
+
+async def get_services(
+    root: "Query", info: "Info", body: typing.Optional[ServiceFilterInput] = None
+) -> typing.List[Service]:
+    db: Session = info.context["db"]
+    stmt = db.query(models.Service)
+
+    if body:
+        if body.id:
+            stmt = stmt.filter(models.Service.id == body.id)
+        elif body.search:
+            stmt = stmt.filter(models.Service.name.like("%" + body.search + "%"))
+
+    services = [service[0] for service in db.execute(stmt).all()]
+    return [Service.from_instance(service) for service in services]
+
+
+async def remove_service(root: "Query", info: "Info", id: int) -> None:
+    db: Session = info.context["db"]
+    db.query(models.Service).filter(models.Service.id == id).delete()
+    db.commit()
+    return None
+
+
+async def add_service(root: "Query", info: "Info", body: AddServiceInput) -> Service:
+    db: Session = info.context["db"]
+    service = models.Service(name=body.name)
+
+    if body.loadbalancer:
+        lb = models.Loadbalancer()
+        for server in body.loadbalancer.servers:
+            lb.servers.append(models.LoadbalancerServers(address=server))
+
+        service.loadbalancer = lb
+
+    db.add(service)
+    db.commit()
+    return Service.from_instance(service)