본문 바로가기
WIKI 기술 지식 베이스

SQLAlchemy + Turso

원문 보기 위키 갱신

SQLAlchemy를 Turso 데이터베이스에 연결하는 방법을 안내해요. libSQL 방언(dialect)을 설치하고 모델을 정의한 뒤, 임베디드 레플리카부터 로컬·인메모리까지 상황별 엔진 구성을 배울 수 있어요.

출처: 문서

본문

시작 전 준비물

시작하기 전에 다음을 확인해 주세요.

단계별 진행

1. SQLAlchemy용 libSQL 방언 설치하기

pip install sqlalchemy-libsql

2. 데이터베이스 자격 증명 가져오기

데이터베이스 URL과 인증 토큰을 준비해 주세요. 공식 문서의 "Retrieve database credentials" 절차를 따르면 돼요. 이 값들은 환경 변수로 보관하는 것이 좋아요.

3. 데이터베이스 모델 만들기

from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

class Base(DeclarativeBase):
    pass

class Foo(Base):
    __tablename__ = "foo"
    id: Mapped[str] = mapped_column(primary_key=True)
    bar: Mapped[str] = mapped_column(String(100))
    def __repr__(self) -> str:
        return f"Item(id={self.id!r}, bar={self.bar!r})"

4. 엔진 만들기

상황에 맞는 구성을 골라 엔진을 만들어요.

Embedded Replicas (임베디드 레플리카)

from dotenv import load_dotenv
from sqlalchemy import create_engine

TURSO_DATABASE_URL = os.environ.get("TURSO_DATABASE_URL")
TURSO_AUTH_TOKEN = os.environ.get("TURSO_AUTH_TOKEN")

engine = create_engine(
     "sqlite+libsql:///embedded.db",
     connect_args={
         "auth_token": TURSO_AUTH_TOKEN,
         "sync_url": TURSO_DATABASE_URL,
     },
)

Remote only (원격 전용)

from dotenv import load_dotenv
from sqlalchemy import create_engine

TURSO_DATABASE_URL = os.environ.get("TURSO_DATABASE_URL")
TURSO_AUTH_TOKEN = os.environ.get("TURSO_AUTH_TOKEN")

engine = create_engine(
    f"sqlite+{TURSO_DATABASE_URL}?secure=true",
    connect_args={
        "auth_token": TURSO_AUTH_TOKEN,
    },
)

Memory Only (인메모리)

from sqlalchemy import create_engine

engine = create_engine("sqlite+libsql://")

Local only (로컬 전용)

from sqlalchemy import create_engine

engine = create_engine("sqlite+libsql:///local.db")

5. 쿼리하기

from sqlalchemy.orm import Session
from sqlalchemy import select
from models import Foo

@app.route("/", methods=(["GET"]))
def home():
    session = Session(engine)

    # get & print foos
    stmt = select(Foo)

    for item in session.scalars(stmt):
        print(item)

더 알아보기 (Learn more)