database.py

import os
from contextlib import contextmanager # contextmanager 임포트
from typing import Generator

from dotenv import load_dotenv

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base, Session # Session 타입 힌트 추가

load_dotenv()

# 환경 변수 로딩 시 변수명 축약 대신 명시적인 이름 사용 권장
MYSQL_USER = os.environ.get("MYSQL_USER")
MYSQL_PASSWORD = os.environ.get("MYSQL_PASSWORD")
MYSQL_HOST = os.environ.get("MYSQL_ALEMBIC_HOST") # alembic 전용 호스트가 아니라면 MYSQL_HOST로 통일 가능
MYSQL_PORT = os.environ.get("MYSQL_ALEMBIC_PORT")
MYSQL_DB = os.environ.get("MYSQL_DATABASE")

# MySQL 연결 설정 (사용자, 비밀번호, 호스트, DB이름 수정)
# 예: mysql+aiomysql://user:password@localhost:3306/dbname
SQLALCHEMY_DATABASE_URL = f"mysql+aiomysql://{MYSQL_USER}:{MYSQL_PASSWORD}@{MYSQL_HOST}:{MYSQL_PORT}/{MYSQL_DB}"
# SQLite 사용 시: SQLALCHEMY_DATABASE_URL = "sqlite:///./myapi.db"

# create_engine 설정: MySQL 연결 시 pool_pre_ping은 좋은 관례입니다.
engine = create_async_engine(
    SQLALCHEMY_DATABASE_URL, 
    echo=True,
    pool_pre_ping=True,  # 연결 유효성 체크 (MySQL 연결 끊김 방지)
    pool_recycle=3600,   # 연결 재사용 시간 설정    
    # SQLite 사용 시: connect_args={"check_same_thread": False}
)

# 세션 생성을 위한 팩토리
async_session = async_sessionmaker(
    bind=engine, 
    class_=AsyncSession, 
    expire_on_commit=False
)


# Base 모델 클래스 생성
Base = declarative_base()

# 참고: Alembic 사용 시 필요한 주석들은 제거하거나 별도의 alembic config 파일로 옮기는 것이 깔끔합니다.

Cargo.toml

# curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
# rustup update
# cargo update

[package]
name = "sheets"
version = "0.1.0"
edition = "2024"
default-run = "sheets"

[[bin]]
name = "sheets"
path = "src/main.rs"

[profile.dev.package.backtrace]
opt-level = 3

[dependencies]
tokio = { version = "1", features = ["full"] }
futures = "0.3"
itertools = "0.12"

serde = "1"
serde_json = "1"

rustls = { version = "0.23", default-features = false, features = ["aws-lc-rs", "logging", "tls12"] }

google-sheets4 = "*"
yup-oauth2 = "12"
Inflector = "0.11.4"

src/main.rs

use std::path::Path;
use tokio::fs::File;
use tokio::io::AsyncWriteExt; // 비동기 쓰기를 위한 트레이트

mod gspread;
use crate::gspread::{SpreadsheetExporter, create_hub};

/// 생성된 SQL 문을 비동기적으로 .sql 파일에 저장
async fn save_sql_to_file_async(sql: &str, file_name: &str) -> tokio::io::Result<()> {
    // 1. 파일 경로 설정
    let path = Path::new(file_name);

    // 2. 비동기 파일 생성 (Tokio fs 사용)
    let mut file = File::create(path).await?;

    // 3. 비동기 데이터 쓰기 (전체 데이터를 안전하게 기록)
    file.write_all(sql.as_bytes()).await?;

    // 4. 버퍼 비우기 및 동기화 (권장 관례)
    file.flush().await?;

    println!("✅ [Async] SQL 파일 저장 완료: {:?}", path.display());
    Ok(())
}

#[tokio::main]
async fn main() -> tokio::io::Result<()> {
    rustls::crypto::aws_lc_rs::default_provider()
        .install_default()
        .expect("Failed to install crypto provider");

    let hub = create_hub("./sheets-xxxxxx-xxxx.json").await;
    let exporter =
        SpreadsheetExporter::new(hub, "xxxxxx-xx-xxxxxx".into());

    let schema = exporter.export_schema().await;
    save_sql_to_file_async(&schema, "output_schema.sql").await?;

    let data = exporter.export_data().await;
    save_sql_to_file_async(&data, "output_data.sql").await?;

    println!("{}\n{}", schema, data);

    Ok(())
}

src/gspread.rs

//extern crate hyper;
//extern crate hyper_rustls;
//extern crate hyper_util;
extern crate google_sheets4 as sheets4;
use sheets4::{Sheets, hyper_rustls, hyper_util, yup_oauth2};

use serde_json::Value;
use std::collections::HashMap;
use futures::future::join_all;
use hyper_util::client::legacy::connect::HttpConnector;
use hyper_rustls::HttpsConnector;

use inflector::Inflector; // 패키지: id = "0.11" (네이밍 변환용)

pub type SheetsHub = Sheets>;

//#[derive(Default, Debug, Clone, PartialEq)]
pub struct SpreadsheetExporter {
    hub: SheetsHub,
    spreadsheet_id: String,
}

impl SpreadsheetExporter {
    pub fn new(hub: SheetsHub, spreadsheet_id: String) -> Self {
        Self { hub, spreadsheet_id }
    }

    // Value 타입을 String으로 안전하게 변환하는 헬퍼 함수
    fn val_to_string(val: &Value) -> String {
        match val {
            Value::String(s) => s.clone(),
            Value::Number(n) => n.to_string(),
            Value::Bool(b) => b.to_string(),
            _ => val.to_string().replace('"', ""),
        }
    }

    // 시트에서 가져온 Vec>를 정제된 HashMap 리스트로 변환
    fn clean_rows(&self, values: &Vec>) -> Vec> {
        if values.is_empty() { return vec![]; }
        let headers: Vec = values[0].iter().map(Self::val_to_string).collect();
        let mut result = Vec::new();

        for row in values.iter().skip(1) {
            if row.is_empty() || Self::val_to_string(&row[0]).trim().is_empty() { break; }
            let mut row_map = HashMap::new();
            for (i, header) in headers.iter().enumerate() {
                let val = row.get(i).map(Self::val_to_string).unwrap_or_default();
                row_map.insert(header.clone(), val);
            }
            result.push(row_map);
        }
        result
    }

    fn split_column(&self, raw_text: &str) -> String {
        raw_text.split(',').map(|s| format!("`{}`", s.trim())).collect::>().join(", ")
    }

    fn format_sql_value(&self, val: &str) -> String {
        let t = val.trim();
        if t.is_empty() || t.to_uppercase() == "NULL" { return "NULL".to_string(); }
        if t.parse::().is_ok() { t.to_string() } else { format!("'{}'", t.replace("'", "''")) }
    }

    async fn export_schema_unit(&self, row: HashMap) -> String {
        let schema = row.get("SCHEMA").cloned().unwrap_or_default();
        let table_name = row.get("TABLE_NAME").cloned().unwrap_or_default();
        let engine = row.get("ENGINE").cloned().unwrap_or_default();
        let auto_increment = row.get("AUTO_INCREMENT").cloned().unwrap_or_default();
        let charset = row.get("CHARSET").cloned().unwrap_or_default();
        let collate = row.get("COLLATE").cloned().unwrap_or_default();
        let ss_id = match row.get("SPREADSHEET_ID") {
            Some(id) => id,
            None => return format!("-- Error: Missing ID for {}", table_name),
        };

        let result = self.hub.spreadsheets().values_batch_get(ss_id)
            .add_ranges("COLUMN!A:Z")
            .add_ranges("INDEX!A:Z")
            .add_ranges("FOREIGN_KEY!A:Z")
            .doit().await;

        let (_, response) = match result {
            Ok(res) => res,
            Err(e) => return format!("-- API Error in {}: {}\n", table_name, e),
        };

        let v_ranges = response.value_ranges.unwrap_or_default();
        let mut all_defs = Vec::new();

        // 1. COLUMN 처리
        if let Some(vals) = v_ranges.get(0).and_then(|r| r.values.as_ref()) {
            for r in self.clean_rows(vals) {
                let f = r.get("FIELD").cloned().unwrap_or_default();
                let t = r.get("TYPE").cloned().unwrap_or_default();
                let s = r.get("SIGNED").cloned().unwrap_or_default();
                let l = r.get("LENGTH").filter(|s| !s.is_empty()).map_or("".to_string(), |v| format!("({})", v));
                let n = if ["NOT", "NO", "FALSE", "0"].contains(&r.get("NULL").unwrap_or(&"".into()).to_uppercase().as_str()) { "NOT NULL" } else { "" };
                let d = r.get("DEFAULT").filter(|s| !s.is_empty()).map_or("".to_string(), |v| format!("DEFAULT '{}'", v.replace("'", "''")));
                let c = r.get("COMMENT").filter(|s| !s.is_empty()).map_or("".to_string(), |v| format!("COMMENT '{}'", v.replace("'", "''")));
                all_defs.push(format!("  `{}` {} {}{} {} {} {} {}", f.to_snake_case(), t, s, l, n, d, r.get("EXTRA").unwrap_or(&"".into()), c).trim().to_string());
            }
        }

        // 2. INDEX 처리
        if let Some(vals) = v_ranges.get(1).and_then(|r| r.values.as_ref()) {
            for r in self.clean_rows(vals) {
                let itype = r.get("TYPE").unwrap_or(&"".into()).to_uppercase();
                let cols = self.split_column(r.get("COLUMN").unwrap_or(&"".into()));
                if itype == "PRIMARY" { all_defs.push(format!("PRIMARY KEY ({})", cols.to_snake_case())); }
                else {
                    let prefix = if itype == "KEY" { "".into() } else { format!("{} ", itype) };
                    all_defs.push(format!("{}KEY `{}` ({})", prefix, r.get("NAME").unwrap_or(&"".into()).to_snake_case(), cols));
                }
            }
        }

        // 3. FOREIGN KEY 처리
        if let Some(vals) = v_ranges.get(2).and_then(|r| r.values.as_ref()) {
            for r in self.clean_rows(vals) {
                let mut fk = format!("CONSTRAINT `{}` FOREIGN KEY (`{}`) REFERENCES `{}` (`{}`)",
                    r.get("NAME").unwrap_or(&"".to_string()).to_snake_case(),
                    r.get("COLUMN").unwrap_or(&"".to_string()).to_snake_case(),
                    r.get("REF_TABLE").unwrap_or(&"".to_string()).to_plural().to_snake_case(),
                    r.get("REF_COLUMN").unwrap_or(&"".to_string()).to_snake_case());
                if let Some(u) = r.get("ON_UPDATE").filter(|s| !s.is_empty()) { fk.push_str(&format!(" ON UPDATE {}", u)); }
                if let Some(d) = r.get("ON_DELETE").filter(|s| !s.is_empty()) { fk.push_str(&format!(" ON DELETE {}", d)); }
                all_defs.push(fk);
            }
        }
        
        let schema_name = schema.to_snake_case();
        let table_name = table_name.to_plural().to_snake_case();        
        format!("CREATE DATABASE IF NOT EXISTS {};\nUSE {};\nDROP TABLE IF EXISTS `{}`;\nCREATE TABLE `{}` (\n{}\n) ENGINE={} AUTO_INCREMENT={} DEFAULT CHARSET={} COLLATE={};\n",
            schema_name, schema_name, table_name, table_name, all_defs.join(",\n"), engine, auto_increment, charset, collate)
    }

    async fn export_data_unit(&self, row: HashMap) -> String {
        let table_name = row.get("TABLE_NAME").cloned().unwrap_or_default();
        let ss_id = row.get("SPREADSHEET_ID").expect("No ID");

        let result = self.hub.spreadsheets().values_get(ss_id, "DATA!A:Z").doit().await;
        let (_, response) = match result {
            Ok(res) => res,
            Err(_) => return "".into(),
        };

        let values = response.values.unwrap_or_default();
        if values.len() <= 1 { return "".into(); }

        let headers: Vec = values[0].iter().map(Self::val_to_string).collect();
        let mut insert_rows = Vec::new();

        for r in self.clean_rows(&values) {
            let row_vals: Vec = headers.iter()
                .map(|h| self.format_sql_value(r.get(h).unwrap_or(&"".into())))
                .collect();
            insert_rows.push(format!("({})", row_vals.join(", ")));
        }

        let h_sql = headers.iter().map(|h| format!("`{}`", h.to_snake_case())).collect::>().join(", ");
        format!("INSERT INTO `{}` ({})\nVALUES\n{};\n", table_name.to_plural(), h_sql, insert_rows.join(",\n"))
    }

    async fn execute_parallel(&self, unit_func: F) -> String 
    where 
        F: Fn(HashMap) -> Fut,
        Fut: std::future::Future
,
    {
        let result = self.hub.spreadsheets().values_get(&self.spreadsheet_id, "SCHEMA!A:Z").doit().await;
        let (_, response) = match result {
            Ok(res) => res,
            Err(e) => return format!("Critical Error: {}", e),
        };

        let values = response.values.unwrap_or_default();
        if values.len() < 2 { return "No data found.".to_string(); }

        let headers: Vec = values[0].iter().map(Self::val_to_string).collect();
        let mut tasks = Vec::new();
        for row in values.iter().skip(1) {
            if !row.is_empty() && Self::val_to_string(&row[0]).to_uppercase().trim() == "TRUE" {
                let mut row_map = HashMap::new();
                for (i, h) in headers.iter().enumerate() {
                    row_map.insert(h.clone(), row.get(i).map(Self::val_to_string).unwrap_or_default());
                }
                tasks.push(unit_func(row_map));
            }
        }
        
        join_all(tasks).await.join("\n")
    }

    pub async fn export_schema(&self) -> String {
        self.execute_parallel(|r| async { self.export_schema_unit(r).await }).await
    }
    pub async fn export_data(&self) -> String {
        self.execute_parallel(|row| async { self.export_data_unit(row).await }).await
    }
}

// 3. 최신 hyper v1 및 hyper-util 기반 인증 설정
pub async fn create_hub(service_account_key: &str) -> SheetsHub {
  // Get an ApplicationSecret instance by some means. It contains the `client_id` and
  // `client_secret`, among other things.
  //parse_service_account_key
  let secret: yup_oauth2::ServiceAccountKey = yup_oauth2::read_service_account_key(service_account_key)
    .await
    .expect("client secret could not be read");
    
  let connector = hyper_rustls::HttpsConnectorBuilder::new()
    .with_native_roots()
    .unwrap()
    .https_or_http()
    .enable_http1()
    .enable_http2()
    .build();

  let client = hyper_util::client::legacy::Client::builder(
    hyper_util::rt::TokioExecutor::new()
    )
    .build(connector);

  let auth = yup_oauth2::ServiceAccountAuthenticator::builder(
  secret,
    )
    .build()
    .await
    .unwrap();

    Sheets::new(client, auth)
}

main.py

from database import async_session
from models.answer import Answer
from models import Question
from sqlalchemy import select, insert
from sqlalchemy.exc import SQLAlchemyError

import asyncio
import SpreadsheetExporter

async def create_user_with_transaction(username: str):
    # 1. 세션 생성
    async with async_session() as session:
        # 2. 트랜잭션 시작 (비동기 컨텍스트 매니저)
        async with session.begin():
            try:
                # 데이터 생성 작업
                new_user = Answer(username=username)
                session.add(new_user)
                
                # 추가 작업 (예: 로그 기록 등)
                # 이 블록 안에서 에러가 발생하면 전체 작업이 롤백됩니다.
                
            except Exception as e:
                # session.begin()을 사용하면 에러 발생 시 자동 롤백되지만, 
                # 추가적인 로깅이 필요하면 여기서 처리합니다.
                print(f"에러 발생: {e}")
                raise

        # 블록을 나가면 자동으로 commit 완료
    print("트랜잭션이 성공적으로 커밋되었습니다.")

async def manual_transaction_example(username: str):
    session = async_session()
    try:
        # 작업 수행
        new_user = Answer(username=username)
        session.add(new_user)
        
        # 명시적 커밋
        await session.commit()
        print("커밋 성공")
    except SQLAlchemyError as e:
        # 에러 발생 시 명시적 롤백
        await session.rollback()
        print(f"롤백 실행: {e}")
    finally:
        # 세션 닫기
        await session.close()


async def create_user_and_post(username: str, title: str):
    async with async_session() as session:
        async with session.begin():
            # 1. 사용자 생성
            user = Answer(username=username)
            session.add(user)
            await session.flush()  # DB에 임시 반영하여 user.id를 확보

            # 2. 해당 사용자의 포스트 생성
            post = Question(title=title, owner_id=user.id)
            session.add(post)
            
            # 여기서 에러 발생 시 User 생성도 취소됨

async def fast_bulk_insert(user_data_list: list):
    async with async_session() as session:
        async with session.begin():
            # Core의 insert 구문 사용
            stmt = insert(Answer).values(user_data_list)
            await session.execute(stmt)

import gspread_asyncio

# from google-auth package
from google.oauth2.service_account import Credentials 

# First, set up a callback function that fetches our credentials off the disk.
# gspread_asyncio needs this to re-authenticate when credentials expire.
def get_creds():
  # To obtain a service account JSON file, follow these steps:
  # https://gspread.readthedocs.io/en/latest/oauth2.html#for-bots-using-service-account

  #path = os.environ.get('GOOGLE_APPLICATION_CREDENTIALS')
  path = "./sheets-xxxxxx-xxxxxxxxxxxx.json"
  creds = Credentials.from_service_account_file(path)
  scoped = creds.with_scopes([
      "https://spreadsheets.google.com/feeds",
      "https://www.googleapis.com/auth/spreadsheets",
      "https://www.googleapis.com/auth/drive",
  ])
  return scoped


if __name__ == "__main__":
  # Execute when the module is not initialized from an import statement.
  #main()

  # --- 사용 예시 ---
  async def main():
    # Create an AsyncioGspreadClientManager object which
    # will give us access to the Spreadsheet API.
    agcm = gspread_asyncio.AsyncioGspreadClientManager(get_creds)

    # Always authorize first.
    # If you have a long-running program call authorize() repeatedly.
    agc = await agcm.authorize()

    spreadsheet_id = "xxxxxxxxx-xxx-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
    exporter = SpreadsheetExporter.SpreadsheetExporter(agc, spreadsheet_id)

    print("--- SQL 스키마 추출 시작 (병렬 처리) ---")
    # 내부적으로 _execute_parallel을 호출하여 SCHEMA 시트의 TRUE 항목을 병렬 처리합니다.
    schema_sql = await exporter.export_schema()
    print(f"{schema_sql}")

    with open("output_schema.sql", "w", encoding="utf-8") as f:
        f.write(schema_sql)
    print("✅ 스키마 추출 완료: output_schema.sql")

    print("\n--- 데이터 INSERT 구문 추출 시작 (병렬 처리) ---")
    # 내부적으로 _execute_parallel을 호출하여 DATA 시트의 TRUE 항목을 병렬 처리합니다.
    data_sql = await exporter.export_data()
    print(f"{data_sql}")

    with open("output_data.sql", "w", encoding="utf-8") as f:
        f.write(data_sql)
    print("✅ 데이터 추출 완료: output_data.sql")
    
  # Turn on debugging if you're new to asyncio!
  asyncio.run(main(), debug=True)    

SpreadsheetExporter.py

import pandas as pd
import numpy as np
import asyncio
from gspread.exceptions import SpreadsheetNotFound, APIError, GSpreadException

class SpreadsheetExporter:
    def __init__(self, agc, spreadsheet_id):
        """
        :param agc: gspread_asyncio의 AsyncioGspreadClient
        :param spreadsheet_id: 메인 SCHEMA 시트 ID
        """
        self.agc = agc
        self.spreadsheet_id = spreadsheet_id

    def _split_column(self, raw_text):
        # 1. 원본 문자열 정의
        #raw_text = "reactable_type, reactable_id, id"
        # 2. 쉼표(,)를 기준으로 분리하여 리스트에 저장 (공백 제거 포함)
        column_list = [item.strip() for item in raw_text.split(',')]

        # 3. 각 요소를 백틱(`)으로 감싼 후 다시 합치기
        result = ", ".join([f"`{item}`" for item in column_list])

        # 4. 결과 출력
        #print(result)
        return result
    
    async def _export_schema_unit(self, row):
        """개별 테이블의 스키마를 생성하는 비동기 단위 작업"""
        try:
            # 구조 분해 할당 (시트 컬럼명에 맞춰 조정 필요)
            table_name = row['TABLE_NAME']
            spreadsheet_id = row['SPREADSHEET_ID']
            engine = row.get('ENGINE', 'InnoDB')
            auto_inc = row.get('AUTO_INCREMENT', '1')
            charset = row.get('CHARSET', 'utf8mb4')
            collate = row.get('COLLATE', 'utf8mb4_unicode_ci')

            ss = await self.agc.open_by_key(spreadsheet_id)
            
            # --- COLUMN, INDEX, FK 시트 데이터를 동시에 가져오기 (병렬) ---
            ws_names = ["COLUMN", "INDEX", "FOREIGN_KEY"]
            worksheets = await asyncio.gather(*[ss.worksheet(name) for name in ws_names])
            data_lists = await asyncio.gather(*[ws.get_all_values() for ws in worksheets])
            
            col_data, idx_data, fk_data = data_lists
            all_defs = []

            # 1. COLUMN 처리
            if len(col_data) > 1:
                df = pd.DataFrame(col_data[1:], columns=col_data[0])

                #df = df[df.iloc[:, 0] != ""].copy()

                """
                # 1. 첫 번째 열이 공백인지 확인 (True/False)
                is_empty = (df.iloc[:, 0] == "")

                # 2. cummax()를 사용하여 한 번 True가 나오면 그 이후는 계속 True가 되도록 함
                # 3. 그 반대(~)인 행들만 남김
                df = df[~is_empty.cummax()].copy()
                """

                # 2. 첫 번째 열에 글자가 '있는지' 확인 (글자 있으면 True, 비었으면 False)
                is_not_empty = (df.iloc[:, 0] != "")

                # 3. cumprod()를 사용하여 한 번 False(공백)를 만나는 순간 그 뒤는 무조건 False로 고정
                # (중간에 첫 칸이 비면 다음 행에 데이터가 있어도 무시하는 핵심 로직)
                df = df[is_not_empty.cumprod().astype(bool)].copy()

                #for row in df.itertuples():
                #print(f"이름: {row.이름}, 점수: {row.점수}")
                type_sql = df.apply(lambda r: f"{r['TYPE']}({r['LENGTH']})" if r.get('LENGTH') else r['TYPE'], axis=1)
                null_sql = df['NULL'].apply(lambda x: "NOT NULL" if str(x).upper() in ["NOT", "NO", "FALSE", "0"] else "")
                default_sql = df['DEFAULT'].apply(lambda x: f"DEFAULT '{str(x).replace("'", "''")}'" if x else "")
                comment_sql = df['COMMENT'].apply(lambda x: f"COMMENT '{str(x)[:1024].replace("'", "''")}'" if x else "")
                
                lines = "  `" + df['FIELD'] + "` " + type_sql + " " + null_sql + " " + default_sql + " " + df['EXTRA'] + " " + comment_sql
                all_defs.extend(lines.str.strip().tolist())

            # 2. INDEX 처리
            if len(idx_data) > 1:
                df = pd.DataFrame(idx_data[1:], columns=idx_data[0])

                #df = df[df.iloc[:, 0] != ""].copy()
                # 1. 첫 번째 열이 공백인지 확인 (True/False)
                is_empty = (df.iloc[:, 0] == "")

                # 2. cummax()를 사용하여 한 번 True가 나오면 그 이후는 계속 True가 되도록 함
                # 3. 그 반대(~)인 행들만 남김
                df = df[~is_empty.cummax()].copy()

                idx_lines = df.apply(lambda r: f"PRIMARY KEY (`{r['COLUMN']}`)" if r['TYPE'].upper() == "PRIMARY" 
                                     #else f"{r['TYPE']+' ' if r['TYPE'].upper()!='KEY' else ''}KEY `{r['NAME']}` (`{r['COLUMN']}`)", axis=1)
                                     else f"{r['TYPE']+' ' if r['TYPE'].upper()!='KEY' else ''}KEY `{r['NAME']}` ({self._split_column(r['COLUMN'])})", axis=1)

                all_defs.extend(idx_lines.tolist())

            # 3. FOREIGN KEY 처리
            if len(fk_data) > 1:
                df = pd.DataFrame(fk_data[1:], columns=fk_data[0])

                #df = df[df.iloc[:, 0] != ""].copy()
                # 1. 첫 번째 열이 공백인지 확인 (True/False)
                is_empty = (df.iloc[:, 0] == "")

                # 2. cummax()를 사용하여 한 번 True가 나오면 그 이후는 계속 True가 되도록 함
                # 3. 그 반대(~)인 행들만 남김
                df = df[~is_empty.cummax()].copy()

                fk_lines = "CONSTRAINT `" + df['NAME'] + "` FOREIGN KEY (`" + df['COLUMN'] + "`) REFERENCES `" + df['REF_TABLE'] + "` (`" + df['REF_COLUMN'] + "`)"
                fk_lines += df['ON_UPDATE'].apply(lambda x: f" ON UPDATE {x}" if x else "")
                fk_lines += df['ON_DELETE'].apply(lambda x: f" ON DELETE {x}" if x else "")
                all_defs.extend(fk_lines.tolist())

            return (f"DROP TABLE IF EXISTS `{table_name}`;\n"
                    f"CREATE TABLE `{table_name}` (\n"
                    f"{',\n'.join(all_defs)}\n"
                    f") ENGINE={engine} AUTO_INCREMENT={auto_inc} DEFAULT CHARSET={charset} COLLATE={collate};\n")
        except Exception as e:
            return f"-- Error in {row.get('TABLE_NAME')}: {e}\n"

    async def _export_data_unit(self, row):
        """개별 테이블의 데이터를 INSERT 문으로 생성하는 비동기 단위 작업"""
        try:
            table_name = row['TABLE_NAME']
            ss = await self.agc.open_by_key(row['SPREADSHEET_ID'])
            ws = await ss.worksheet("DATA")
            data = await ws.get_all_values()
            
            if len(data) <= 1: return ""

            df = pd.DataFrame(data[1:], columns=data[0])

            #df = df[df.iloc[:, 0] != ""].copy()
            # 1. 첫 번째 열이 공백인지 확인 (True/False)
            is_empty = (df.iloc[:, 0] == "")

            # 2. cummax()를 사용하여 한 번 True가 나오면 그 이후는 계속 True가 되도록 함
            # 3. 그 반대(~)인 행들만 남김
            df = df[~is_empty.cummax()].copy()

            # 데이터 벡터화 처리 (Pandas)
            for col in df.columns:
                df[col] = df[col].astype(str).str.replace("'", "''")
                mask = ~df[col].str.match(r'^-?\d+(\.\d+)?$') # 숫자가 아니면 따옴표
                df.loc[mask, col] = "'" + df[col].loc[mask] + "'"
                df.loc[df[col] == "''", col] = "NULL"

            values_sql = "(" + df.agg(', '.join, axis=1) + ")"
            header = ", ".join([f"`{h}`" for h in data[0]])
            return f"INSERT INTO `{table_name}` ({header})\nVALUES\n{',\n'.join(values_sql)};\n"
        except Exception as e:
            return f"-- Data Error in {row.get('TABLE_NAME')}: {e}\n"

    async def export_schema(self):
        """메인 실행 함수: 모든 테이블 스키마 병렬 처리"""
        return await self._execute_parallel(self._export_schema_unit)

    async def export_data(self):
        """메인 실행 함수: 모든 테이블 데이터 병렬 처리"""
        return await self._execute_parallel(self._export_data_unit)

    async def _execute_parallel(self, unit_func):
        """SCHEMA 시트를 읽고 병렬로 태스크를 실행하는 공통 로직"""
        try:
            main_ss = await self.agc.open_by_key(self.spreadsheet_id)
            schema_ws = await main_ss.worksheet("SCHEMA")
            raw_data = await schema_ws.get_all_values()
            
            if len(raw_data) < 2: return "No data found."

            df_main = pd.DataFrame(raw_data[1:], columns=raw_data[0])

            #df = df[df.iloc[:, 0] != ""].copy()
            # 1. 첫 번째 열이 공백인지 확인 (True/False)
            is_empty = (df_main.iloc[:, 0] == "")

            # 2. cummax()를 사용하여 한 번 True가 나오면 그 이후는 계속 True가 되도록 함
            # 3. 그 반대(~)인 행들만 남김
            df_main = df_main[~is_empty.cummax()].copy()

            # CHK 컬럼이 TRUE인 대상만 선별
            active_targets = df_main[df_main.iloc[:, 0].str.upper().str.strip() == "TRUE"]

            # --- 핵심: asyncio.gather를 통한 병렬 실행 ---
            tasks = [unit_func(row) for _, row in active_targets.iterrows()]
            #tasks = [unit_func(row) for row in active_targets.itertuples()]
            results = await asyncio.gather(*tasks)
            
            return "\n".join(results)
        except Exception as e:
            return f"Critical Error: {e}"


"""
⚡ 최적화 포인트 설명
계층적 병렬화 (Double asyncio.gather):
상위 레벨: _execute_parallel에서 여러 테이블(Table A, Table B...)을 동시에 처리합니다.
하위 레벨: _export_schema_unit 내부에서 한 테이블의 COLUMN, INDEX, FK 시트 3개를 동시에 읽어옵니다. (기존 대비 시트 로딩 속도 약 3배 향상)
Pandas 벡터화 연산: for 루프 없이 수천 줄의 INSERT 값이나 DDL 구문을 한 번에 생성합니다.
메모리 효율: df[df.iloc[:, 0] != ""] 코드를 통해 구글 시트 하단의 불필요한 빈 행을 즉시 제거하여 메모리 낭비를 방지합니다.
안정성: asyncio.gather 도중 하나의 시트에서 에러가 발생해도 전체 프로세스가 죽지 않도록 개별 유닛 함수에 try-except를 배치하여 에러 메시지를 SQL 주석(-- Error...) 형태로 반환하게 했습니다.
이제 이 코드를 사용하면 구글 API 할당량(Quota) 내에서 가장 빠른 속도로 SQL 추출이 가능합니다. gspread-asyncio 공식 문서를 참고하여 AsyncioGspreadClient를 넘겨주시면 바로 작동합니다.
"""

"""
"reactable_type, reactable_id, id" 문자열을 , 로 분리후 리스트에 저장한후 
"`reactable_type`, `reactable_id`, `id`" 형식으로 화면에 출력해주는 파이썬 코드를 생성해줘

# 1. 원본 문자열 정의
raw_text = "reactable_type, reactable_id, id"

# 2. 쉼표(,)를 기준으로 분리하여 리스트에 저장 (공백 제거 포함)
column_list = [item.strip() for item in raw_text.split(',')]

# 3. 각 요소를 백틱(`)으로 감싼 후 다시 합치기
result = ", ".join([f"`{item}`" for item in column_list])

# 4. 결과 출력
print(result)
"""

app/domain/sheet/__init__.py

# domain/sheet/__init__.py

# 하위 파일들(router, schemas, crud)을 패키지 레벨로 끌어올립니다.
from . import crud, router, schemas

# 외부에서 'from domain.answer import *'를 하거나 가져갈 수 있는 범위를 명시합니다.
__all__ = ["router", "schemas", "crud"]