%%sql
DROP TABLE IF EXISTS items;

CREATE TABLE IF NOT EXISTS
items (
    id INT PRIMARY KEY,
    name VARCHAR(255),
    price FLOAT
);

import logging

from sqlalchemy import create_engine, text
import requests

ca_cert_url = "https://portal.singlestore.com/static/ca/singlestore_bundle.pem"
ca_cert_path = "/tmp/singlestore_bundle.pem"

response = requests.get(ca_cert_url)
with open(ca_cert_path, "wb") as f:
    f.write(response.content)

sql_connection_string = connection_url.replace("singlestoredb", "mysql+pymysql")

engine = create_engine(
    f"{sql_connection_string}?ssl_ca={ca_cert_path}",
    pool_size=10,           # Maximum number of connections in the pool is 10
    max_overflow=5,         # Allow up to 5 additional connections (temporary overflow)
    pool_timeout=30         # Wait up to 30 seconds for a connection from the pool
)
logging.info("Connection to workspace established successfully.")


def execute_query(query: str):
    logging.info(f"Executing query: {query}")
    with engine.connect() as connection:
        return connection.execute(text(query))

def execute_transaction(transactional_query: str):
    logging.info(f"Starting transaction for query: {transactional_query}")
    with engine.connect() as connection:
        transaction = connection.begin()
        try:
            result = connection.execute(text(transactional_query))
            transaction.commit()
            logging.info("Transaction committed successfully.")
            return result
        except Exception as e:
            transaction.rollback()
            logging.error(f"Transaction rolled back due to error: {e}")
            raise e

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from singlestoredb import connect
from concurrent.futures import ThreadPoolExecutor
import asyncio

# Define the Type of the Data
class Item(BaseModel):
    id: int
    name: str
    price: float

# Create an executor that can execute queries on multiple threads simultaneously
executor = ThreadPoolExecutor()
def run_in_thread(fn, *args):
    loop = asyncio.get_event_loop()
    return loop.run_in_executor(executor, fn, *args)

app = FastAPI()

# add logging middleware
@app.middleware("http")
async def log_requests(request, call_next):
    logging.info(f"Incoming request: {request.method} {request.url}")
    response = await call_next(request)
    logging.info(f"Response status: {response.status_code}")
    return response

# Get all items
@app.get("/items", response_model=list[Item])
async def get_items():
    def get_items_query():
        result = execute_query("SELECT * FROM items;")
        rows = result.fetchall()
        logging.info(f"Fetched {len(rows)} items from the database")
        return [{"id": row[0], "name": row[1], "price": row[2]} for row in rows]

    try:
        return await run_in_thread(get_items_query)
    except Exception as e:
        logging.error(f"Error fetching all items: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error fetching all items: {str(e)}")

# Insert an item
@app.post("/items", response_model=dict)
async def create_item(item: Item):
    def insert_item_query():
        logging.info(f"Inserting item: {item}")
        result = execute_transaction(f"INSERT INTO items (id, name, price) VALUES ({item.id}, '{item.name}', {item.price})")
        logging.info(f"Item with id {item.id} inserted successfully")
        return {"message": f"Item with id {item.id} inserted successfully"}

    try:
        return await run_in_thread(insert_item_query)
    except Exception as e:
        logging.error(f"Error while inserting item with id {item.id}: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error while inserting item with id {item.id}: {str(e)}")

# Get item by id
@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: int):
    def get_item_query():
        result = execute_query(f"SELECT * FROM items WHERE id={item_id}")
        row = result.fetchone()
        if not row:
            logging.error(f"Item with id {item_id} not found")
            raise HTTPException(status_code=404, detail="Item not found")
        logging.info(f"Item with id {item_id} fetched successfully")
        return {"id": row[0], "name": row[1], "price": row[2]}

    try:
        return await run_in_thread(get_item_query)
    except HTTPException as e:
        raise e
    except Exception as e:
        logging.error(f"Error fetching item with id {item_id}: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error fetching item with id {item_id}: {str(e)}")

# Delete item by id
@app.delete("/items/{item_id}", response_model=dict)
async def delete_item(item_id: int):
    logging.info(f"Deleting item with id {item_id}")
    def delete_item_query():
        result = execute_transaction(f"DELETE FROM items WHERE id={item_id}")
        logging.info(f"Number of rows deleted: {result.rowcount}")
        return {"message": f"number of rows deleted: {result.rowcount}"}

    try:
        return await run_in_thread(delete_item_query)
    except Exception as e:
        logging.error(f"Error deleting item with id {item_id}: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error deleting item with id {item_id}: {str(e)}")

import singlestoredb.apps as apps
connection_info = await apps.run_function_app(app)