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 from xmlrpc.client import ServerProxy def connect(): return create_engine("postgresql+psycopg://postgres:Root12345678@10.10.8.83:32101/db", echo=True) def exec_function(function: str, **kwargs): proxy = ServerProxy('http://10.10.8.70:7000/xmlrpc') result = proxy.__getattr__(function)(kwargs) return result 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] upstream: Mapped[bool] profiles: Mapped[List['Profile']] = relationship( back_populates='user', cascade='all, delete-orphan' ) schedule: Mapped[List['Schedule']] = 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, 'upstream': self.upstream, 'profiles': [x.to_dict() for x in self.profiles], 'schedule': [x.to_dict() for x in self.schedule], } @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, } class Schedule(Base): __tablename__ = 'schedule' id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column(ForeignKey('users.id')) day_start: Mapped[int] hour_start: Mapped[int] minute_start: Mapped[int] day_end: Mapped[int] hour_end: Mapped[int] minute_end: Mapped[int] user: Mapped['User'] = relationship(back_populates='schedule') def to_dict(self) -> dict: return { 'id': self.id, 'day_start': self.day_start, 'hour_start': self.hour_start, 'minute_start': self.minute_start, 'day_end': self.day_end, 'hour_end': self.hour_end, 'minute_end': self.minute_end, } @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 update_schedule(self, id_: int, day_start: int, day_end: int, hour_start: int, hour_end: int, minute_start: int, minute_end: int) -> bool: sync_engine = connect() with Session(sync_engine) as session: schedule = session.query(Schedule).get(id_) schedule.day_start = day_start schedule.day_end = day_end schedule.hour_start = hour_start schedule.hour_end = hour_end schedule.minute_start = minute_start schedule.minute_end = minute_end session.commit() return True @strawberry.mutation def create_schedule(self, user_id: int, day_start: int, day_end: int, hour_start: int, hour_end: int, minute_start: int, minute_end: int) -> int: sync_engine = connect() with (Session(sync_engine) as session): user = session.query(User).get(user_id) if not user: return 0 schedule = Schedule(user_id=user_id, day_start=day_start, day_end=day_end, hour_start=hour_start, hour_end=hour_end, minute_start=minute_start, minute_end=minute_end) session.add_all([schedule]) session.commit() return schedule.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 @strawberry.mutation def remove_schedule(self, id_: int) -> bool: sync_engine = connect() with (Session(sync_engine) as session): schedule = session.query(Schedule).get(id_) if schedule: session.delete(schedule) 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, upstream=False ) bnd128 = User( username="bnd128", passwd="gost_2012$a742ec53198ec2a5027086fba8814a89982a57112d1a72d02260161108f39b50", bndname="bnd128", newbnd=True, active=True, upstream=False ) session.add_all([bnd127, bnd128]) session.commit() def main(): init() uvicorn.run("main:app", port=9000, log_level="info") schema = strawberry.Schema(query=Query, mutation=Mutation) graphql_app = GraphQLRouter(schema, graphiql=True) app = FastAPI(root_path="/graphql") 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()