schema.py

import strawberry
import os
from dotenv import load_dotenv
import requests
from typing import List
import gradio as gr
import multiprocessing

load_dotenv()

FLOWISE_URL = os.getenv('FLOWISE_URL')
HEADERS = {'Authorization': os.getenv('FLOWISE_TOKEN')}
API_KEY = os.getenv('OPENWEATHER_API_KEY')

# schema and resolver about weather
@strawberry.type
class Wind:
    speed: float
    deg: int

@strawberry.type
class Clouds:
    all: int

@strawberry.type
class Weather:
    city: str
    main: str
    description: str
    temperature: float
    feelsLike: float
    tempMin: float
    tempMax: float
    pressure: int
    humidity: float
    seaLevel: int
    wind: Wind
    clouds: Clouds

# weather API call and data mapping
def getWeather(city: str) -> Weather:
    url = f'http://api.openweathermap.org/data/2.5/weather?q={city}&appid={API_KEY}&units=metric'
    response = requests.get(url)
    data = response.json()

    wind = Wind(speed=data['wind']['speed'], deg=data['wind']['deg'])
    clouds = Clouds(all=data['clouds']['all'])

    weather = Weather(
        city=data['name'],
        main=data['weather'][0]['main'],
        description=data['weather'][0]['description'],
        temperature=data['main']['temp'],
        feelsLike=data['main']['feels_like'],
        tempMin=data['main']['temp_min'],
        tempMax=data['main']['temp_max'],
        pressure=data['main']['pressure'],
        humidity=data['main']['humidity'],
        seaLevel=data['main'].get('sea_level', 0),
        wind=wind,
        clouds=clouds
    )

    return weather

# datetime API call and data mapping
def getCurrentDay(timezone: str)-> str:
    url = f'https://worldtimeapi.org/api/timezone/{timezone}'
    response = requests.get(url)
    data = response.json()
    return data['datetime']

# schema and resolver about flowise
# Function to connect to the Flowise API   
def connectFlowise(message: str, history: List[str] = None) -> str:
    if history is None:
        history = []
    payload = {'question': message}
    try:
        response = requests.post(FLOWISE_URL, json=payload, headers=HEADERS)
        response.raise_for_status()
        return response.json()['text']
    except requests.exceptions.RequestException:
        return "Sorry!, I cann't connect to the server right now. Please, try again later."

# Function to connect to the Gradio Interface  
def connectGradio(port:int, server:str):
    # Header personalitation
    header = """
    <h1 style="text-align: center;">🌤️ Your Personal Weather Assistant 🌤️</h1>
    """        
    interface = gr.ChatInterface(
        connectFlowise,
        theme="soft",
        chatbot=gr.Chatbot(placeholder="<strong>Your Personal Weather Assistant</strong><br>You can ask me about the weather or anything else related"),
        title="Chatbot Assistant",
        fill_height=True,
        description=header
    )
    def launch_interface():
        interface.launch(server_port=port, server_name=server)

    # Get away gradio process from current graph ql server
    process = multiprocessing.Process(target=launch_interface)
    process.start()
    return f"Application running on {server}:{port}"

@strawberry.type
class Query:
    weather: Weather = strawberry.field(resolver=getWeather)
    date: str = strawberry.field(resolver=getCurrentDay)
    flowise: str = strawberry.field(resolver=connectFlowise)
    gradio: str = strawberry.field(resolver=connectGradio)

schema = strawberry.Schema(query=Query)