106 lines
2.6 KiB
Python
106 lines
2.6 KiB
Python
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)
|