728x90
728x90
Postgresql을 psycopg2를 사용해 연동하기
supabase를 사용하다 보니 기존에서 사용하던 서버에서 supabase의 postgresql과 연동할 일이 생겨서
기회가 생긴 김에 포스팅을 하려고 합니다
이번 포스팅은 선행사항으로 python으로 FAST API로 서버를 생성한 상태입니다
database.py 설정
pip install psycopg2 명령어를 사용하여 패키지를 설치해 줍니다
파일에 DB를 연결하고 사용하기 위한 초기 설정을 해줍니다
✍ postgres_url = postgresql의 URL을 받는 부분
✍ postgres_db_name = DB의 이름을 받는 부분
그러나두 개의 정보는 다른 사람들에게 보이면 보안에 문제가 되는 코드이기에
ENV(환경변수)로 저장을 해줍니다
💡 python에서 env를 설정하는 방법은 다음 포스팅에서 다루도록 해보겠습니다
# 📲 postgresql을 연결할때 민감한 여러번 적을 필요 없이 한번에 적기 위한 코드
connection_string = "{}/{}".format(postgres_url, postgres_db_name)
class PostgresDB():
# 📲 클래스의 인스턴스가 생성될 때 실행되는 초기화 함수
def __init__(self):
# 📲 postgresql에 연결하는 부분
self.db = psycopg2.connect(connection_string)
# 📲 cursor은 SQL 쿼리를 실행하는데 사용
self.cursor = self.db.cursor()
# 📲 클래스의 인스턴스가 사라질 때 실행되는 함수
# 📲 postgresql과 cursor를 닫습니다
def __del__(self):
self.db.close()
self.cursor.close()
# 📲 SQL 쿼리를 실행하는 함수
def execute(self,query,args={}):
self.cursor.execute(query,args)
# 📲 결과물을 담는 코드
row = self.cursor.fetchall()
return row
# 📲 cursor에 대한 모든 변경사항을 commit합니다
def commit(self):
self.cursor.commit()
DB 정보를 Model화 시킨 후 CRUD
Pydantic을 사용해서 모델을 정의해 주면
데이터타입을 헷갈리지 않고 디버깅에도 유용하기에 모델로 만들어줍니다
그 뒤에 SQLAlchemy를 사용하여 ORM으로 쿼리를 직접 작성하지 않고도 복잡한 쿼리를 수행가능하게 해 줍니다
물론 여기에서는 CRUD처럼 간단한 쿼리만 진행하지만 해두면 손해는 없습니다 ^^
from sqlalchemy import create_engine, Table, MetaData
from sqlalchemy.orm import sessionmaker
from sqlalchemy import desc
class UserModel(BaseModel):
id : str
name : Optional[str]
age : Optional[str]
memo : Optional[str] = ''
class UserData():
def __init__(self):
# 📲 엔진을 초기화 하고 세션을 생성하는 부분
engine = create_engine(connection_string)
Session = sessionmaker(bind=engine)
self.session = Session()
# 📲 user 테이블을 메타데이터로 읽어오는 부분
metadata = MetaData()
self.table = Table('user', metadata, autoload_with=engine)
def insert(self, data: SoundModel):
# 📲 받아온 데이터를 딕셔너리로 변환 후 insert 수행하는 부분
self.new_entry = data.dict()
self.session.execute(self.table.insert(), self.new_entry)
self.session.commit()
def read(self):
# 📲 테이블에 대한 쿼리 생성
query = self.session.query(self.table)
# 📲 테이블의 id 컬럼으로 순서를 정렬해주는 부분
query = query.order_by(desc(self.table.columns.id))
rows = query.all()
result = [row._asdict() for row in rows]
return result
def update(self, id: str, data: UserModel):
# 📲 데이터를 딕셔너리 형태로 변환합니다.
# 📲 exclude_unset는 값이 설정되지 않은 필드들은 딕셔너리에서 제외하는 옵션
update_data = data.dict(exclude_unset=True)
self.session.execute(
self.table.update().where(self.table.c.id == id).values(**update_data)
)
self.session.commit()
def delete(self, id: str):
#📲 특정 사용자를 삭제하는 쿼리를 실행하는 부분
self.session.execute(
self.table.delete().where(self.table.c.id == id)
)
self.session.commit()
CRUD 사용 예시
이제 CRUD를 사용하는 것을 예제로 통해 알아보겠습니다
이때의 데이터를 집어넣는 부분은 유연하게 테스트해보셔도 됩니다
from database import UserData, UserModel
<<INSERT>>
# 📲 insert할 데이터를 넣는 부분
# 📲 이때 memo는 Optional 이기 때문에 오류가 발생하지 않습니다
async def dataInsert():
userInsert = UserModel(
id= '1',
name= 'nuridal',
age= '10')
# 📲 실질적으로 insert가 이루어 지는 부분
UserData().insert(userInsert)
<<READ>>
# 📲 user 테이블의 데이터를 모두 가져오는 함수
async def dataRead():
status = UserData().read()
return status
<<UPDATE>>
# 📲 인자로 준 id가 1일 경우에 업데이트를 수행하는 부분
async def dataUpdate():
UserData().update(id='1',
data=UserModel(
id= '1',
age='20',
memo= 'test age update'))
<<DELETE>>
#📲 삭제하려는 사용자의 id로 delete가 수행되는 부분
async def dataDelete():
id = '1'
UserData().delete(id)오늘은 Python으로 Postgresql을 연동하는 방법에 대해서 알아보았습니다
조금이나마 도움이 되었으면 좋겠습니다
그럼 코딩이 쉬워지는 그날까지!!!
728x90
300x250