<div id="readability-page-1" class="page"><div>
<h2 data-ke-size="size26">개요</h2>
<p data-ke-size="size16">FastAPI + SQLModel Postgres DB연동 검색해도 잘 안나와서 포스팅 합니다.</p>

<p data-ke-size="size16">다음 공식 사이트는 sqlite로 되어 있습니다.</p>
<p data-ke-size="size16"><a href="https://sqlmodel.tiangolo.com/" target="_blank" rel="noopener&nbsp;noreferrer">https://sqlmodel.tiangolo.com/</a></p>
<figure id="og_1708311950848" contenteditable="false" data-ke-type="opengraph" data-ke-align="alignCenter" data-og-type="website" data-og-title="SQLModel" data-og-description="SQLModel, SQL databases in Python, designed for simplicity, compatibility, and robustness." data-og-host="sqlmodel.tiangolo.com" data-og-source-url="https://sqlmodel.tiangolo.com/" data-og-url="https://sqlmodel.tiangolo.com/" data-og-image="https://scrap.kakaocdn.net/dn/bYk2KR/hyVmQG0hb2/qmzYsMrEjWNiCKRhJZTDH0/img.png?width=1200&amp;height=630&amp;face=0_0_1200_630,https://scrap.kakaocdn.net/dn/zvl7U/hyVi9nUgWU/7WnMYaEECqvUc9pXk1aOtk/img.png?width=1200&amp;height=630&amp;face=0_0_1200_630,https://scrap.kakaocdn.net/dn/JqDYR/hyVmYZj1pT/3igOzBtKpjvXVGtMKXNdak/img.png?width=730&amp;height=540&amp;face=0_0_730_540"><a href="https://sqlmodel.tiangolo.com/" target="_blank" rel="noopener" data-source-url="https://sqlmodel.tiangolo.com/">

<div>
<p data-ke-size="size16">SQLModel</p>
<p data-ke-size="size16">SQLModel, SQL databases in Python, designed for simplicity, compatibility, and robustness.</p>
<p data-ke-size="size16">sqlmodel.tiangolo.com</p>
</div>
</a><img src="https://scrap.kakaocdn.net/dn/bYk2KR/hyVmQG0hb2/qmzYsMrEjWNiCKRhJZTDH0/img.png?width=1200&amp;height=630&amp;face=0_0_1200_630,https://scrap.kakaocdn.net/dn/zvl7U/hyVi9nUgWU/7WnMYaEECqvUc9pXk1aOtk/img.png?width=1200&amp;height=630&amp;face=0_0_1200_630,https://scrap.kakaocdn.net/dn/JqDYR/hyVmYZj1pT/3igOzBtKpjvXVGtMKXNdak/img.png?width=730&amp;height=540&amp;face=0_0_730_540"></figure>


<p data-ke-size="size16">postgres연동으로 교체한 후 실행</p>
<div>
<pre data-ke-language="java"><code>from typing import Optional

from sqlmodel import Field, SQLModel, create_engine, Session, select


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None


hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)

# DB 접속 정보
db_host = "your_db_host" # ex) ec2-123-123-123-123.ap-northeast-2.compute.amazonaws.com
user = "your_db_user"
password = "your_db_password"

engine = create_engine(f'postgresql://{user}:{password}@{db_host}/your_db_name')


SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    session.add(hero_1)
    session.add(hero_2)
    session.add(hero_3)
    session.commit()

with Session(engine) as session:
    statement = select(Hero).where(Hero.name == "Spider-Boy")
    hero = session.exec(statement).first()
    print(hero)</code></pre>
</div>

<h4 data-ke-size="size20">접속정보 env처리</h4>
<p data-ke-size="size16">보안상 소스코드에 DB접속정보는 넣지 않습니다. 다음과 같이 .env처리 할 수 있습니다.</p>
<div>
<pre><code># DB 접속 정보
db_host = os.getenv('DB_HOST')
user = os.getenv('DB_USERNAME')
password = os.getenv('DB_PASSWORD')</code></pre>
</div>


<h3 data-ke-size="size23">결과</h3>
<h4 data-ke-size="size20">DB에 생성</h4>
<figure data-ke-mobilestyle="widthOrigin" data-origin-width="392" data-origin-height="289"><span data-url="https://blog.kakaocdn.net/dn/UtsZY/btsEYM9UERM/5h6ydyud8robmrXTKA8xW0/img.png" data-lightbox="lightbox"><img src="https://blog.kakaocdn.net/dn/UtsZY/btsEYM9UERM/5h6ydyud8robmrXTKA8xW0/img.png" onerror="this.onerror=null; this.src='//t1.daumcdn.net/tistory_admin/static/images/no-image-v1.png'; this.srcset='//t1.daumcdn.net/tistory_admin/static/images/no-image-v1.png';" data-origin-width="392" data-origin-height="289"></span></figure>


<figure data-ke-mobilestyle="widthOrigin" data-origin-width="736" data-origin-height="206"><span data-url="https://blog.kakaocdn.net/dn/bb0eFt/btsEY8rmO6b/1eSi32OMJFHZKDMejqGXMk/img.png" data-lightbox="lightbox"><img src="https://blog.kakaocdn.net/dn/bb0eFt/btsEY8rmO6b/1eSi32OMJFHZKDMejqGXMk/img.png" onerror="this.onerror=null; this.src='//t1.daumcdn.net/tistory_admin/static/images/no-image-v1.png'; this.srcset='//t1.daumcdn.net/tistory_admin/static/images/no-image-v1.png';" data-origin-width="736" data-origin-height="206"></span></figure>


<h4 data-ke-size="size20">콘솔 출력</h4>
<figure data-ke-mobilestyle="widthOrigin" data-origin-width="1181" data-origin-height="197"><span data-url="https://blog.kakaocdn.net/dn/w8oD6/btsEYKxGTkF/oisBYkmCJnfF30MnoNgjUK/img.png" data-lightbox="lightbox"><img src="https://blog.kakaocdn.net/dn/w8oD6/btsEYKxGTkF/oisBYkmCJnfF30MnoNgjUK/img.png" onerror="this.onerror=null; this.src='//t1.daumcdn.net/tistory_admin/static/images/no-image-v1.png'; this.srcset='//t1.daumcdn.net/tistory_admin/static/images/no-image-v1.png';" data-origin-width="1181" data-origin-height="197"></span></figure>



<h2 data-ke-size="size26">DB Connection분리</h2>

<p data-ke-size="size16">session.py</p>
<div>
<pre><code>import os

from sqlmodel import create_engine, Session, SQLModel

db_host = os.getenv('DB_HOST')
user = os.getenv('DB_USERNAME')
password = os.getenv('DB_PASSWORD')

engine = create_engine(f'postgresql://{user}:{password}@{db_host}')


def get_session() -&gt; Session:
    with Session(engine) as session:
        yield session


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)</code></pre>
</div>

<p data-ke-size="size16">test_db_integration.py</p>
<p data-ke-size="size16">위 session.py에서 yield session으로 리턴 했기 때문에 next(get_session())을 써서 session 객체를 받아줍니다.</p>
<div>
<pre><code>from sqlmodel import select

from db.session import get_session, create_db_and_tables
from models.user_model import User


create_db_and_tables()

hero_1 = User(name="Kyeongrok", secret_name="Dive Wilson", age=38)
hero_2 = User(name="Taehwan", secret_name="Pedro Parqueador")
hero_3 = User(name="Junbin", secret_name="Tommy Sharp")

session = next(get_session())  # Generator를 썼기 때문에 next()를 사용해야 합니다.
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()

statement = select(User).where(User.name == "Deadpond")
hero = session.exec(statement).first()
print(hero)
</code></pre>
</div>

<h2 data-ke-size="size26">Router적용</h2>
<p data-ke-size="size16">main.py에 너무 많은 내용이 들어있기 때문에 router를 이용해 endpoint를 분리 해줍니다.</p>
<figure data-ke-mobilestyle="widthOrigin" data-origin-width="326" data-origin-height="607"><span data-url="https://blog.kakaocdn.net/dn/bRyDqk/btsEZAPekJE/HUMN0T5EwaZixbAoXzaCNK/img.png" data-lightbox="lightbox"><img src="https://blog.kakaocdn.net/dn/bRyDqk/btsEZAPekJE/HUMN0T5EwaZixbAoXzaCNK/img.png" onerror="this.onerror=null; this.src='//t1.daumcdn.net/tistory_admin/static/images/no-image-v1.png'; this.srcset='//t1.daumcdn.net/tistory_admin/static/images/no-image-v1.png';" data-origin-width="326" data-origin-height="607"></span></figure>

<p data-ke-size="size16">main.py</p>
<div>
<pre><code>from fastapi import FastAPI

from api.v1.api import api_router

app = FastAPI(
    title="Flow View API",
    openapi_url="/api/v1/openapi.json"
)

app.include_router(api_router)</code></pre>
</div>

<p data-ke-size="size16">api.v1/api.py</p>
<div>
<pre><code>from fastapi import APIRouter

from api.v1.endpoints import (root, user)

api_router = APIRouter()
api_router.include_router(root.router, tags=["root"])
api_router.include_router(user.router, prefix="/users", tags=["users"])
</code></pre>
</div>


<p data-ke-size="size16">api.v1.endpoints/root.py</p>
<div>
<pre><code>from fastapi import APIRouter

router = APIRouter()


@router.get("/")
async def root():
    return {"message": "Hello World"}


@router.get("/hello/{name}")
async def say_hello(name: str):
    # return apple string
    return {"message": f"Hello {name}"}
</code></pre>
</div></div></div>

뷰티풀 프로그래밍

개요
FastAPI + SQLModel Postgres DB연동 검색해도 잘 안나와서 포스팅 합니다.
 
다음 공식 사이트는 sqlite로 되어 있습니다.
https://sqlmodel.tiangolo.com/
 
Postgres연동
postgres연동으로 교체한 후 실행
from typing import Optional
from sqlmodel import Field, SQLModel, create_engine, Session, select
class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)
# DB 접속 정보
db_host = "your_db_host" # ex) ec2-123-123-123-123.ap-northeast-2.compute.amazonaws.com
user = "your_db_user"
password = "your_db_password"
engine = create_engine(f'postgresql://{user}:{password}@{db_host}/your_db_name')
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
    session.add(hero_1)
    session.add(hero_2)
    session.add(hero_3)
    session.commit()
with Session(engine) as session:
    statement = select(Hero).where(Hero.name == "Spider-Boy")
    hero = session.exec(statement).first()
    print(hero)
 
접속정보 env처리
보안상 소스코드에 DB접속정보는 넣지 않습니다. 다음과 같이 .env처리 할 수 있습니다.
# DB 접속 정보
db_host = os.getenv('DB_HOST')
user = os.getenv('DB_USERNAME')
password = os.getenv('DB_PASSWORD')
 
 
결과
DB에 생성
 
 
콘솔 출력
 
 
DB Connection분리
 
session.py
import os
from sqlmodel import create_engine, Session, SQLModel
db_host = os.getenv('DB_HOST')
user = os.getenv('DB_USERNAME')
password = os.getenv('DB_PASSWORD')
engine = create_engine(f'postgresql://{user}:{password}@{db_host}')
def get_session() -> Session:
    with Session(engine) as session:
        yield session
def create_db_and_tables():
    SQLModel.metadata.create_all(engine)
 
test_db_integration.py
위 session.py에서 yield session으로 리턴 했기 때문에 next(get_session())을 써서 session 객체를 받아줍니다.
from sqlmodel import select
from db.session import get_session, create_db_and_tables
from models.user_model import User
create_db_and_tables()
hero_1 = User(name="Kyeongrok", secret_name="Dive Wilson", age=38)
hero_2 = User(name="Taehwan", secret_name="Pedro Parqueador")
hero_3 = User(name="Junbin", secret_name="Tommy Sharp")
session = next(get_session())  # Generator를 썼기 때문에 next()를 사용해야 합니다.
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()
statement = select(User).where(User.name == "Deadpond")
hero = session.exec(statement).first()
print(hero)
 
Router적용
main.py에 너무 많은 내용이 들어있기 때문에 router를 이용해 endpoint를 분리 해줍니다.
main.py
from fastapi import FastAPI
from api.v1.api import api_router
app = FastAPI(
    title="Flow View API",
    openapi_url="/api/v1/openapi.json"
)
app.include_router(api_router)
 
api.v1/api.py
from fastapi import APIRouter
from api.v1.endpoints import (root, user)
api_router = APIRouter()
api_router.include_router(root.router, tags=["root"])
api_router.include_router(user.router, prefix="/users", tags=["users"])
 
 
api.v1.endpoints/root.py
from fastapi import APIRouter
router = APIRouter()
@router.get("/")
async def root():
    return {"message": "Hello World"}
@router.get("/hello/{name}")
async def say_hello(name: str):
    # return apple string
    return {"message": f"Hello {name}"}

FastAPI + SQLModel Postgres DB연동

댓글