SQLAlchemy Metadata로 테이블, 쿼리 개발. JOIN 샘플

SQLAlchemy에서 Table Join을 복잡하게 하는 경우가 많아, 가급적 쉬운 방법을 찾아보았다.

 

전체소스 코드 : https://github.com/armyost/hexagonalSample/tree/main

GitHub - armyost/hexagonalSample: This is python Sample with Hexagonal Architecture

 

1-1) src/app/domain/models/user.py 에 Class 모델 정의

class User:

    def __init__(self, userId, description, userName, deptId):
        self.userId = userId
        self.description = description
        self.userName = userName
        self.deptId = deptId

    def __repr__(self):
        return f'<User {self.userId}>'

    def as_dict(self):
        return {
            'userId': self.userId,
            'description': self.description,
            'userName': self.userName,
            'deptId': self.deptId
        }

1-2) src/app/domain/models/department.py 에 Class 모델 정의

class Department:

    def __init__(self, deptId, description, deptName):
        self.deptId = deptId
        self.description = description
        self.deptName = deptName

    def __repr__(self):
        return f'<Department {self.deptId}>'

    def as_dict(self):
        return {
            'deptId': self.deptId,
            'description': self.description,
            'deptName': self.deptName
        }

 

2-1) src/app/domain/repositories/user_repository_interface.py 에 다음과 같이 인터페이스 정의

from abc import (
    ABCMeta,
    abstractmethod
)

class UserRepositoryInterface:
    __metaclass__ = ABCMeta

    @abstractmethod
    def insert(self, user):
        pass

    @abstractmethod
    def selectWithDeptInfo(self, userId):
        pass

 

2-2) src/app/domain/repositories/department_repository_interface.py 에 다음과 같이 인터페이스 정의

from abc import (
    ABCMeta,
    abstractmethod
)

class DepartmentRepositoryInterface:
    __metaclass__ = ABCMeta

    @abstractmethod
    def insert(self, department):
        pass

 

3) src/app/infrastructure/database.py 에 다음과 같이 정의

users_table = Table(
    'user', metadata,
    Column('userId', Integer, primary_key=True),
    Column('description', Text, nullable=True),
    Column('userName', String(10), nullable=True),
    Column('deptId', Integer, ForeignKey("department.deptId"), nullable=True)
)

departments_table = Table(
    'department', metadata,
    Column('deptId', Integer, primary_key=True),
    Column('description', Text, nullable=True),
    Column('deptName', String(10), nullable=True)
)

 

※ MetaData 형식의 테이블 매핑 정의서 만드는 링크

https://docs.sqlalchemy.org/en/20/core/metadata.html

Describing Databases with MetaData — SQLAlchemy 2.0 Documentation

 

 

 

 

4-1) src/app/infrastructure/repositories/user_repository.py 에서 UserRepository 구현하고,

class UserRepository(BaseRepository, UserRepositoryInterface):

    def insert(self, user):
        with self.db_connection.begin():
            self.db_connection.execute(
                users_table.insert(),
                user.as_dict()
            )

    def selectWithDeptInfo(self, userId):
        with self.db_connection.begin():
            stmt = select(users_table, departments_table).outerjoin(departments_table, users_table.c.deptId == departments_table.c.deptId).where(users_table.c.userId == userId)
            row = self.db_connection.execute(stmt).fetchone()
            return {
                'userId' : row['userId'],
                'userName' : row['userName'],
                'deptName' : row['deptName']
            } if row else None

 

4-2) src/app/infrastructure/repositories/department_repository.py 에서 DepartmentRepository 구현한다.

class DepartmentRepository(BaseRepository, DepartmentRepositoryInterface):

    def insert(self, department):
        with self.db_connection.begin():
            self.db_connection.execute(
                departments_table.insert(),
                department.as_dict()
            )

※ SQLAlchemy의 Query정의 문서 : https://docs.sqlalchemy.org/en/20/orm/queryguide/select.html