itv-config/main.py

from typing import List, Optional
from sqlalchemy import create_engine, String, select, ForeignKey
from sqlalchemy.orm import Session, DeclarativeBase, Mapped, mapped_column, relationship
import strawberry
import json
import uvicorn
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
from starlette.middleware.cors import CORSMiddleware
from strawberry.scalars import JSON


def connect():
    return create_engine("postgresql+psycopg://postgres:Root12345678@10.10.8.83:32101/db", echo=True)


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str]
    passwd: Mapped[str]
    bndname: Mapped[str]
    newbnd: Mapped[bool]
    active: Mapped[bool]

    profiles: Mapped[List["Profile"]] = relationship(
        back_populates="user", cascade="all, delete-orphan"
    )

    def __repr__(self) -> str:
        return f"User(id={self.id!r}, username={self.username!r}, password={self.passwd!r}, newbnd={self.newbnd})"

    def to_dict(self) -> dict:
        return {
            'id': self.id,
            'username': self.username,
            'bndname': self.bndname,
            'newbnd': self.newbnd,
            'active': self.active,
            'profiles': [x.to_dict() for x in self.profiles]
        }

@strawberry.type
class UsersGQL:
    users: JSON


class Profile(Base):
    __tablename__ = "profiles"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    scheme: Mapped[str]
    json: Mapped[str]

    user: Mapped["User"] = relationship(back_populates="profiles")

    def to_dict(self) -> dict:
        return {
            'id': self.id,
            'user_id': self.user_id,
            'json': self.json,
        }


@strawberry.type
class Query:
    @strawberry.field()
    def users(self) -> UsersGQL:
        sync_engine = connect()
        with Session(sync_engine) as session:
            stmt = select(User)
            return UsersGQL(users=[json.dumps(user.to_dict()) for user in session.scalars(stmt)])


@strawberry.type
class Mutation:
    @strawberry.mutation
    def update_user(self, id_: int, bndname: str, newbnd: bool, active: bool) -> bool:
        sync_engine = connect()
        with Session(sync_engine) as session:
            user = session.query(User).get(id_)
            user.bndname = bndname
            user.newbnd = newbnd
            user.active = active
            session.commit()
        return True

    @strawberry.mutation
    def create_user(self, username: str, passwd: str) -> int:
        sync_engine = connect()
        with Session(sync_engine) as session:
            user = User(username=username, passwd=passwd, bndname=username, active=False, newbnd=True)
            session.add_all([user])
            session.commit()
            return user.id

    @strawberry.mutation
    def update_profile(self, id_: int, scheme: str, json: str) -> bool:
        sync_engine = connect()
        with Session(sync_engine) as session:
            profile = session.query(Profile).get(id_)
            profile.scheme = scheme
            profile.json = json
            session.commit()
        return True

    @strawberry.mutation
    def create_profile(self, user_id: int, scheme: str, json: str) -> int:
        sync_engine = connect()
        with (Session(sync_engine) as session):
            user = session.query(User).get(user_id)
            if not user:
                return 0
            profile = Profile(user_id=user_id, scheme=scheme, json=json)
            session.add_all([profile])
            session.commit()
            return profile.id

    @strawberry.mutation
    def remove_user(self, id_: int) -> bool:
        sync_engine = connect()
        with (Session(sync_engine) as session):
            user = session.query(User).get(id_)
            if user:
                session.delete(user)
                session.commit()
        return True

    @strawberry.mutation
    def remove_profile(self, id_: int) -> bool:
        sync_engine = connect()
        with (Session(sync_engine) as session):
            profile = session.query(Profile).get(id_)
            if profile:
                session.delete(profile)
                session.commit()
        return True


def init():
    sync_engine = connect()
    Base.metadata.drop_all(sync_engine)
    Base.metadata.create_all(sync_engine)
    with Session(sync_engine) as session:
        bnd127 = User(
            username="bnd127",
            passwd="gost_2012$a742ec53198ec2a5027086fba8814a89982a57112d1a72d02260161108f39b50",
            bndname="bnd127",
            newbnd=True,
            active=True,
        )
        bnd128 = User(
            username="bnd128",
            passwd="gost_2012$a742ec53198ec2a5027086fba8814a89982a57112d1a72d02260161108f39b50",
            bndname="bnd128",
            newbnd=True,
            active=True,
        )
        session.add_all([bnd127, bnd128])
        session.commit()


def main():
    uvicorn.run("main:app", port=9000, log_level="info")


schema = strawberry.Schema(query=Query, mutation=Mutation)

graphql_app = GraphQLRouter(schema, graphiql=True)

app = FastAPI()
app.add_middleware(CORSMiddleware,
                   allow_origins=['http://127.0.0.1'],
                   allow_credentials=True,
                   allow_methods=['*'],
                   allow_headers=['*']
                   )
app.include_router(graphql_app, prefix='/graphql')

if __name__ == '__main__':
    main()