Flama를 사용한 기본 도메인 기반 디자인-파이썬 튜토리얼-php.cn

ML API의 개발 및 생산화에 도움이 되는 몇 가지 흥미로운 새 기능을 제공하는 Flama 1.7의 최근 릴리스에 대해 이미 들어보셨을 것입니다. 이 게시물은 해당 릴리스의 주요 특징 중 하나인 도메인 기반 디자인 지원에 대해 자세히 설명합니다. 그러나 실제 사례를 통해 세부 사항을 살펴보기 전에 다음 리소스를 염두에 두는 것이 좋습니다(아직 숙지하지 않았다면 숙지하시기 바랍니다).

Flama 공식 문서: Flama 문서
ML API용 Flama 소개 게시물: 강력한 기계 학습 API용 Flama 소개

이제 새로운 기능을 시작하고 이를 활용하여 강력하고 유지 관리가 가능한 ML API를 구축하는 방법을 살펴보겠습니다.

도메인 중심 디자인이란 무엇입니까?

간략한 개요

현대 소프트웨어 개발에서는 비즈니스 로직을 애플리케이션의 기술 설계에 맞추는 것이 필수적입니다. 이것이 도메인 중심 설계(DDD)가 빛을 발하는 곳입니다. DDD는 비즈니스의 핵심 영역을 반영하는 소프트웨어 구축을 강조하고, 비즈니스 개념을 중심으로 코드를 구성하여 복잡한 문제를 해결합니다. 이를 통해 DDD는 개발자가 유지 관리 가능하고 확장 가능하며 강력한 응용 프로그램을 만들 수 있도록 도와줍니다. 다음에서는 여러분이 알아야 할 DDD의 가장 중요한 개념을 소개합니다. 이에 대해 자세히 알아보기 전에 이 게시물은 DDD에 대한 포괄적인 가이드도 아니고 해당 주제에 대한 주요 참고 자료를 대체할 의도도 없다는 점을 말씀드리고 싶습니다. 실제로 DDD에 대해 더 깊이 이해하려면 다음 리소스를 권장합니다.

Cosmic Python(Harry Percival 및 Bob Gregory 저): 이 책은 Python에서 DDD를 적용하는 방법을 배울 수 있는 훌륭한 리소스입니다.
Domain-Driven Design: Tackling Complexity in the Heart of Software(Eric Evans 저): 이 책은 DDD를 세상에 소개한 책으로, DDD에 대한 깊은 이해를 키우는 데 관심이 있는 사람이라면 반드시 읽어야 할 책입니다.

주요 개념

DDD의 주요 개념을 자세히 알아보기 전에 Cosmic Python의 매우 유용한 그림을 살펴보는 것이 좋습니다. 이 그림에서는 이러한 개념이 앱의 맥락에서 표시되어 서로 어떻게 연결되어 있는지 보여줍니다. .

도메인 모델

도메인 모델의 개념은 간단한 용어 정의로 설명할 수 있습니다.

도메인은 당사 소프트웨어가 지원하기 위해 구축되고 있는 활동(또는 지식)의 특정 주제 영역을 나타냅니다.
모델은 소프트웨어에서 인코딩하려는 시스템이나 프로세스의 간단한 표현(또는 추상화)을 의미합니다.

따라서 도메인 모델은 비즈니스 소유자가 비즈니스 운영 방식에 대해 염두에 두고 있는 일련의 개념과 규칙을 참조하는 멋진(그러나 표준적이고 유용한) 방법입니다. 이는 시스템의 동작을 제어하는 규칙, 제약 조건 및 관계를 포함하여 우리가 흔히 애플리케이션의 비즈니스 로직이라고 부르는 것입니다.

이제부터 도메인 모델을 모델이라고 부르겠습니다.

저장소 패턴

리포지토리 패턴은 모델과 데이터 액세스를 분리할 수 있는 디자인 패턴입니다. 리포지토리 패턴의 기본 아이디어는 데이터 액세스 논리와 애플리케이션의 비즈니스 논리 사이에 추상화 계층을 만드는 것입니다. 이 추상화 계층을 사용하면 문제를 분리하여 코드를 더욱 유지 관리하고 테스트할 수 있습니다.

저장소 패턴을 구현할 때 일반적으로 다른 저장소에서 구현해야 하는 표준 메서드를 지정하는 인터페이스(AbstractRepository)를 정의합니다. 그런 다음 데이터 액세스 논리가 구현되는 이러한 메서드의 구체적인 구현으로 특정 저장소가 정의됩니다(예: SQLAlchemyRepository). 이 디자인 패턴은 데이터 조작 방법을 분리하여 애플리케이션의 다른 곳에서 원활하게 사용할 수 있도록 하는 것을 목표로 합니다. 우리 도메인 모델에서.

작업 패턴 단위

작업 패턴 단위는 데이터 액세스에서 모델을 최종적으로 분리하는 데 누락된 부분입니다. 작업 단위는 데이터 액세스 논리를 캡슐화하고 단일 트랜잭션 내에서 데이터 소스에 대해 수행되어야 하는 모든 작업을 그룹화하는 방법을 제공합니다. 이 패턴을 사용하면 모든 작업이 원자적으로 수행됩니다.

작업 단위 패턴을 구현할 때 일반적으로 다른 작업 단위가 구현해야 하는 표준 메서드(AbstractUnitOfWork)를 지정하는 인터페이스를 정의합니다. 그런 다음 데이터 액세스 논리가 구현되는 이러한 메서드의 구체적인 구현을 통해 특정 작업 단위가 정의됩니다(예: SQLAlchemyUnitOfWork). 이 디자인을 사용하면 애플리케이션의 비즈니스 로직 구현을 변경할 필요 없이 데이터 소스에 대한 연결을 체계적으로 처리할 수 있습니다.

Flama로 DDD 구현하기

DDD의 주요 개념을 간략하게 소개한 후 Flama를 사용하여 DDD를 구현하는 방법을 알아볼 준비가 되었습니다. 이 섹션에서는 Flama를 사용하여 개발 환경을 설정하고, 기본 애플리케이션을 구축하고, DDD 개념을 구현하는 과정을 안내합니다.

예를 진행하기 전에 방금 검토한 주요 DDD 개념과 관련된 Flama의 명명 규칙을 살펴보세요.

Native Domain-Driven Design with Flama

위 그림에서 볼 수 있듯이 명명 규칙은 매우 직관적입니다. Repository는 저장소 패턴을 나타냅니다. 그리고, Worker는 작업 단위를 의미합니다. 이제 DDD를 사용하는 Flama API 구현으로 넘어갈 수 있습니다. 하지만 시작하기 전에 flama를 사용하여 간단한 API를 만드는 방법이나 이미 코드가 준비된 후 API를 실행하는 방법에 대한 기본 사항을 검토해야 한다면 확인해 보는 것이 좋습니다. 빠른 시작 가이드를 살펴보세요. 여기에서 이 게시물을 수행하는 데 필요한 기본 개념과 단계를 찾을 수 있습니다. 이제 더 이상 고민하지 말고 구현을 시작해 보겠습니다.

개발 환경 설정

첫 번째 단계는 개발 환경을 만들고 이 프로젝트에 필요한 모든 종속성을 설치하는 것입니다. 좋은 점은 이 예에서는 JWT 인증을 구현하는 데 필요한 모든 도구를 갖기 위해 flama만 설치하면 된다는 것입니다. 종속성을 관리하기 위해 시를 사용할 예정이지만 원하는 경우 pip를 사용할 수도 있습니다.

poetry add "flama[full]" "aiosqlite"

로그인 후 복사

이 예제에서 사용할 데이터베이스인 SQLAlchemy와 함께 SQLite를 사용하려면 aiosqlite 패키지가 필요합니다.

우리가 일반적으로 프로젝트를 어떻게 구성하는지 알고 싶다면 여기에서 이전 게시물을 살펴보세요. 여기서는 시가 포함된 Python 프로젝트를 설정하는 방법과 우리가 일반적으로 따르는 프로젝트 폴더 구조에 대해 자세히 설명합니다.

기본 응용 프로그램

단일 공개 엔드포인트가 있는 간단한 애플리케이션부터 시작해 보겠습니다. 이 엔드포인트는 API에 대한 간략한 설명을 반환합니다.

# src/app.py
from flama import Flama

app = Flama(
    title="Domain-driven API",
    version="1.0.0",
    description="Domain-driven design with Flama ?",
    docs="/docs/",
)

@app.get("/", name="info")
def info():
    """
    tags:
        - Info 
    summary:
        Ping
    description:
        Returns a brief description of the API
    responses:
        200:
            description:
                Successful ping.
    """
    return {"title": app.schema.title, "description": app.schema.description, "public": True}

로그인 후 복사

이 애플리케이션을 실행하려면 위의 코드를 src 폴더 아래 app.py라는 파일에 저장한 후 다음 명령을 실행하면 됩니다. (시 환경을 활성화해야 한다는 점을 기억하세요. 그렇지 않으면 다음 작업을 수행해야 합니다.) 명령 앞에 시 실행을 붙입니다):

flama run --server-reload src.app:app

INFO:     Started server process [3267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

로그인 후 복사

여기서 --server-reload 플래그는 선택 사항이며 코드가 변경될 때 서버를 자동으로 다시 로드하는 데 사용됩니다. 이는 개발 중에 매우 유용하지만 필요하지 않은 경우 제거할 수 있습니다. 사용 가능한 옵션의 전체 목록을 보려면 flama run --help를 실행하거나 설명서를 확인하세요.

또는 다음 스크립트를 실행하여 애플리케이션을 실행할 수도 있습니다. 이 스크립트는 src 폴더 아래에 __main__.py로 저장할 수 있습니다.

# src/__main__.py
import flama

def main():
    flama.run(
      flama_app="src.app:app", 
      server_host="0.0.0.0", 
      server_port=8000, 
      server_reload=True
    )

if __name__ == "__main__":
    main()

로그인 후 복사

그런 다음 다음 명령을 실행하여 애플리케이션을 실행할 수 있습니다.

poetry add "flama[full]" "aiosqlite"

로그인 후 복사

DDD 실행 중

이제 애플리케이션에 대한 최소한의 뼈대를 설정했으므로
내에서 방금 검토한 DDD 개념 구현을 시작할 수 있습니다. 실제 시나리오를 모방하려는 간단한 예의 맥락입니다. 사용자 관리를 위한 API 개발을 요청받았고, 다음과 같은 요구 사항이 제공되었다고 가정해 보겠습니다.

/user/에 대한 POST 요청을 통해 사용자의 이름, 성, 이메일 및 비밀번호를 제공하여 새 사용자를 생성하려고 합니다.
생성된 모든 사용자는 다음 스키마를 사용하여 데이터베이스에 저장됩니다.
- id: 사용자의 고유 식별자입니다.
- 이름: 사용자 이름.
- 성: 사용자의 성.
- 이메일: 사용자의 이메일입니다.
- 비밀번호: 사용자의 비밀번호입니다. 데이터베이스에 저장하기 전에 해시해야 합니다.
- active: 사용자가 활성 상태인지 여부를 나타내는 부울 플래그입니다. 기본적으로 사용자는 비활성 상태로 생성됩니다.
생성된 사용자는 이메일과 비밀번호를 사용하여 /user/activate/에 POST 요청을 보내 계정을 활성화해야 합니다. 사용자가 활성화되면 사용자의 상태가 데이터베이스에서 활성으로 업데이트되어야 합니다.
사용자는 이메일과 비밀번호를 사용하여 /user/signin/에 POST 요청을 보내 로그인할 수 있습니다. 사용자가 활성 상태인 경우 API는 모든 사용자 정보를 반환해야 합니다. 그렇지 않으면 API가 오류 메시지를 반환해야 합니다.
계정을 비활성화하려는 사용자는 이메일과 비밀번호를 포함하여 /user/deactivate/에 POST 요청을 보내면 됩니다. 사용자가 비활성화되면 데이터베이스에서 사용자 상태를 비활성으로 업데이트해야 합니다.

이 요구 사항 집합은 이전에 애플리케이션의 도메인 모델이라고 불렀던 것으로 구성되며, 이는 기본적으로 다음 사용자 워크플로를 구체화한 것에 지나지 않습니다.

/user/에 대한 POST 요청을 통해 사용자가 생성됩니다.
사용자는 /user/activate/에 대한 POST 요청을 통해 계정을 활성화합니다.
사용자는 /user/signin/에 대한 POST 요청을 통해 로그인합니다.
사용자가 /user/deactivate/에 대한 POST 요청을 통해 계정을 비활성화합니다.
사용자는 2~4단계를 원하는 만큼 반복할 수 있습니다.

이제 저장소와 작업자 패턴을 사용하여 도메인 모델을 구현해 보겠습니다. 먼저 데이터 모델을 정의한 다음 저장소와 작업자 패턴을 구현하겠습니다.

데이터 모델

사용자 데이터는 SQLite 데이터베이스에 저장됩니다(SQLAlchemy에서 지원하는 다른 데이터베이스를 사용할 수 있습니다). 다음 데이터 모델을 사용하여 사용자를 나타냅니다(이 코드를 src 폴더 아래의 models.py 파일에 저장할 수 있음).

# src/app.py
from flama import Flama

app = Flama(
    title="Domain-driven API",
    version="1.0.0",
    description="Domain-driven design with Flama ?",
    docs="/docs/",
)

@app.get("/", name="info")
def info():
    """
    tags:
        - Info 
    summary:
        Ping
    description:
        Returns a brief description of the API
    responses:
        200:
            description:
                Successful ping.
    """
    return {"title": app.schema.title, "description": app.schema.description, "public": True}

로그인 후 복사

데이터 모델 외에도 데이터베이스와 테이블을 생성하려면 마이그레이션 스크립트가 필요합니다. 이를 위해 프로젝트 루트에 있는 migrations.py라는 파일에 다음 코드를 저장할 수 있습니다.

poetry add "flama[full]" "aiosqlite"

로그인 후 복사

그런 다음 다음 명령을 실행하여 마이그레이션 스크립트를 실행할 수 있습니다.

# src/app.py
from flama import Flama

app = Flama(
    title="Domain-driven API",
    version="1.0.0",
    description="Domain-driven design with Flama ?",
    docs="/docs/",
)

@app.get("/", name="info")
def info():
    """
    tags:
        - Info 
    summary:
        Ping
    description:
        Returns a brief description of the API
    responses:
        200:
            description:
                Successful ping.
    """
    return {"title": app.schema.title, "description": app.schema.description, "public": True}

로그인 후 복사

저장소

이 예에서는 하나의 저장소, 즉 사용자 테이블의 원자적 작업을 처리할 저장소(이름은 UserRepository)만 필요합니다. 다행히 flama에서는 SQLAlchemyTableRepository라는 SQLAlchemy 테이블과 관련된 저장소에 대한 기본 클래스를 제공합니다.

SQLAlchemyTableRepository 클래스는 테이블에서 CRUD 작업을 수행하는 메서드 집합을 제공합니다. 특히 다음과 같습니다.

create: 테이블에 새 요소를 생성합니다. 요소가 이미 존재하는 경우 예외(IntegrityError)가 발생하고, 그렇지 않으면 새 요소의 기본 키가 반환됩니다.
검색: 테이블에서 요소를 검색합니다. 요소가 존재하지 않으면 예외(NotFoundError)가 발생하고, 그렇지 않으면 요소를 반환합니다. 요소가 두 개 이상 발견되면 예외(MultipleRecordsError)가 발생합니다.
업데이트: 테이블의 요소를 업데이트합니다. 요소가 존재하지 않으면 예외(NotFoundError)가 발생하고, 그렇지 않으면 업데이트된 요소가 반환됩니다.
삭제: 테이블에서 요소를 삭제합니다.
목록: 전달된 절 및 필터와 일치하는 테이블의 모든 요소를 나열합니다. 절이나 필터가 제공되지 않으면 테이블의 모든 요소가 반환됩니다. 요소가 발견되지 않으면 빈 목록을 반환합니다.
drop: 데이터베이스에서 테이블을 삭제합니다.

이 예에서는 테이블에 대한 추가 작업이 필요하지 않으므로 SQLAlchemyTableRepository에서 제공하는 메서드로 충분합니다. src 폴더 아래 repositories.py라는 파일에 다음 코드를 저장할 수 있습니다.

flama run --server-reload src.app:app

INFO:     Started server process [3267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

로그인 후 복사

보시다시피 UserRepository 클래스는 SQLAlchemyTableRepository의 하위 클래스이며 _table 속성에 테이블을 설정하기만 하면 됩니다. 이것이 사용자 테이블에 대해 완전한 기능을 갖춘 저장소를 갖기 위해 수행해야 하는 유일한 작업입니다.

표준 CRUD 작업 외에 사용자 정의 메서드를 추가하려면 UserRepository 클래스에서 해당 메서드를 정의하면 됩니다. 예를 들어 활성 사용자 수를 계산하는 방법을 추가하려면 다음과 같이 할 수 있습니다.

# src/__main__.py
import flama

def main():
    flama.run(
      flama_app="src.app:app", 
      server_host="0.0.0.0", 
      server_port=8000, 
      server_reload=True
    )

if __name__ == "__main__":
    main()

로그인 후 복사

이 예에서는 이 방법을 사용하지 않지만 필요한 경우 저장소에 사용자 정의 방법을 추가할 수 있다는 점과 구현 방법을 알아두면 좋습니다.
저장소 패턴의 맥락에서. 이는 애플리케이션의 비즈니스 로직(해당 리소스 메소드에서 구현됨)을 변경하지 않고도 여기에서 모든 데이터 액세스 로직을 구현할 수 있기 때문에 이미 볼 수 있듯이 강력한 디자인 패턴입니다.

노동자

작업 단위 패턴은 데이터 액세스 논리를 캡슐화하고 단일 트랜잭션 내에서 데이터 소스에 대해 수행되어야 하는 모든 작업을 그룹화하는 방법을 제공하는 데 사용됩니다. flama에서는 UoW 패턴이 Worker라는 이름으로 구현됩니다. 저장소 패턴과 마찬가지로 flama는 SQLAlchemyWorker라는 SQLAlchemy 테이블과 관련된 작업자를 위한 기본 클래스를 제공합니다. 본질적으로 SQLAlchemyWorker는 데이터베이스에 대한 연결과 트랜잭션을 제공하고 작업자 연결을 통해 모든 저장소를 인스턴스화합니다. 이 예에서 작업자는 단일 저장소(즉, UserRepository)만 사용하지만 필요한 경우 더 많은 저장소를 추가할 수 있습니다.

우리 작업자는 RegisterWorker라고 하며, src 폴더 아래에workers.py라는 파일에 다음 코드를 저장할 수 있습니다.

poetry add "flama[full]" "aiosqlite"

로그인 후 복사

따라서 ProductRepository 및 OrderRepository와 같이 작업할 저장소가 더 있는 경우 다음과 같이 작업자에 추가할 수 있습니다.

# src/app.py
from flama import Flama

app = Flama(
    title="Domain-driven API",
    version="1.0.0",
    description="Domain-driven design with Flama ?",
    docs="/docs/",
)

@app.get("/", name="info")
def info():
    """
    tags:
        - Info 
    summary:
        Ping
    description:
        Returns a brief description of the API
    responses:
        200:
            description:
                Successful ping.
    """
    return {"title": app.schema.title, "description": app.schema.description, "public": True}

로그인 후 복사

그렇게 간단하게 우리는 애플리케이션에 저장소와 작업자 패턴을 구현했습니다. 이제 사용자 데이터와 상호 작용하는 데 필요한 API 엔드포인트를 제공하는 리소스 메서드를 구현할 수 있습니다.

자원

리소스는 flama 애플리케이션의 주요 구성 요소 중 하나입니다. 이는 애플리케이션 리소스(RESTful 리소스라는 의미)를 나타내고 이들과 상호 작용하는 API 엔드포인트를 정의하는 데 사용됩니다.

이 예에서는 사용자를 생성, 활성화, 로그인 및 비활성화하는 방법이 포함된 UserResource라는 사용자 리소스를 정의합니다. 리소스는 최소한 flama 내장 Resource 클래스에서 파생되어야 합니다. 단, flama는 RESTResource 및 CRUDResource와 같이 작업할 수 있는 보다 정교한 클래스를 제공합니다.

src 폴더 아래 resources.py라는 파일에 다음 코드를 저장할 수 있습니다.

flama run --server-reload src.app:app

INFO:     Started server process [3267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

로그인 후 복사

DDD를 사용한 기본 애플리케이션

이제 데이터 모델, 저장소 및 작업자 패턴, 리소스 메서드를 구현했으므로 이전에 소개한 기본 애플리케이션을 수정하여 모든 것이 예상대로 작동하도록 해야 합니다. 우리는 다음을 수행해야 합니다.

애플리케이션에 SQLAlchemy 연결을 추가하면 SQLAlchemyModule을 애플리케이션 생성자에 모듈로 추가하면 됩니다.
애플리케이션에 작업자를 추가하려면 RegisterWorker를 애플리케이션 생성자에 구성요소로 추가하면 됩니다.

app.py 파일은 다음과 같이 남습니다.

poetry add "flama[full]" "aiosqlite"

로그인 후 복사

DDD 패턴을 통해 애플리케이션의 비즈니스 로직(리소스 메소드에서 쉽게 읽을 수 있음)을 데이터 액세스 로직 (저장소 및 작업자 패턴에서 구현됨) 또한 이러한 우려 사항의 분리로 인해 코드의 유지 관리 및 테스트가 어떻게 더 용이해졌는지, 그리고 코드가 이 예의 시작 부분에서 제시한 비즈니스 요구 사항에 어떻게 더 잘 부합하는지 주목할 가치가 있습니다.

애플리케이션 실행

명령을 실행하기 전에 개발 환경이 올바르게 설정되었는지, 폴더 구조가 다음과 같은지 확인하세요.

# src/app.py
from flama import Flama

app = Flama(
    title="Domain-driven API",
    version="1.0.0",
    description="Domain-driven design with Flama ?",
    docs="/docs/",
)

@app.get("/", name="info")
def info():
    """
    tags:
        - Info 
    summary:
        Ping
    description:
        Returns a brief description of the API
    responses:
        200:
            description:
                Successful ping.
    """
    return {"title": app.schema.title, "description": app.schema.description, "public": True}

로그인 후 복사

모든 것이 올바르게 설정되면 다음 명령을 실행하여 애플리케이션을 실행할 수 있습니다(애플리케이션을 실행하기 전에 마이그레이션 스크립트를 실행하는 것을 잊지 마세요).

flama run --server-reload src.app:app

INFO:     Started server process [3267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

로그인 후 복사

이제 방금 구현한 비즈니스 로직을 시험해 볼 수 있습니다. 컬이나 Postman과 같은 도구를 사용하거나 브라우저에서 http://localhost:8000/docs/로 이동하여

flama에서 제공하는 자동 생성 문서 UI를 사용하여 이 작업을 시도할 수 있습니다. 거기에서 엔드포인트를 시도합니다.

Native Domain-Driven Design with Flama

사용자 생성

사용자를 생성하려면 다음 페이로드를 사용하여 /user/에 POST 요청을 보낼 수 있습니다.

# src/__main__.py
import flama

def main():
    flama.run(
      flama_app="src.app:app", 
      server_host="0.0.0.0", 
      server_port=8000, 
      server_reload=True
    )

if __name__ == "__main__":
    main()

로그인 후 복사

따라서 컬을 사용하여 다음과 같이 요청을 보낼 수 있습니다.

poetry run python src/__main__.py

INFO:     Started server process [3267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

로그인 후 복사

요청이 성공하면 본문이 비어 있는 200 응답을 받아야 하며 데이터베이스에 사용자가 생성됩니다.

로그인

로그인하려면 다음 페이로드를 사용하여 /user/signin/에 POST 요청을 보낼 수 있습니다.

# src/models.py
import uuid

import sqlalchemy
from flama.sqlalchemy import metadata
from sqlalchemy.dialects.postgresql import UUID

__all__ = ["user_table", "metadata"]

user_table = sqlalchemy.Table(
    "user",
    metadata,
    sqlalchemy.Column("id", UUID(as_uuid=True), primary_key=True, nullable=False, default=uuid.uuid4),
    sqlalchemy.Column("name", sqlalchemy.String, nullable=False),
    sqlalchemy.Column("surname", sqlalchemy.String, nullable=False),
    sqlalchemy.Column("email", sqlalchemy.String, nullable=False, unique=True),
    sqlalchemy.Column("password", sqlalchemy.String, nullable=False),
    sqlalchemy.Column("active", sqlalchemy.Boolean, nullable=False),
)

로그인 후 복사

따라서 컬을 사용하여 다음과 같이 요청을 보낼 수 있습니다.

# migrations.py
from sqlalchemy import create_engine

from src.models import metadata

if __name__ == "__main__":
    # Set up the SQLite database
    engine = create_engine("sqlite:///models.db", echo=False)

    # Create the database tables
    metadata.create_all(engine)

    # Print a success message
    print("Database and User table created successfully.")

로그인 후 복사

사용자가 활성 상태가 아닌 경우 다음과 같은 응답을 받아야 합니다.

> poetry run python migrations.py

Database and User table created successfully.

로그인 후 복사

누군가 잘못된 비밀번호로 로그인을 시도하면 어떻게 되는지 테스트할 수도 있습니다.

# src/repositories.py
from flama.ddd import SQLAlchemyTableRepository

from src import models

__all__ = ["UserRepository"]

class UserRepository(SQLAlchemyTableRepository):
    _table = models.user_table

로그인 후 복사

이 경우 다음 본문과 함께 401 응답을 받아야 합니다.

# src/repositories.py
from flama.ddd import SQLAlchemyTableRepository

from src import models

__all__ = ["UserRepository"]

class UserRepository(SQLAlchemyTableRepository):
    _table = models.user_table

    async def count_active_users(self):
        return len((await self._connection.execute(self._table.select().where(self._table.c.active == True))).all())

로그인 후 복사

마지막으로 존재하지 않는 사용자로 로그인을 시도해야 합니다.

# src/workers.py
from flama.ddd import SQLAlchemyWorker

from src import repositories

__all__ = ["RegisterWorker"]


class RegisterWorker(SQLAlchemyWorker):
    user: repositories.UserRepository

로그인 후 복사

이 경우 다음 본문과 함께 404 응답을 받아야 합니다.

# src/workers.py
from flama.ddd import SQLAlchemyWorker

from src import repositories

__all__ = ["RegisterWorker"]

class RegisterWorker(SQLAlchemyWorker):
    user: repositories.UserRepository
    product: repositories.ProductRepository
    order: repositories.OrderRepository

로그인 후 복사

사용자 활성화

로그인 프로세스를 살펴본 후 이제 사용자의 자격 증명을 사용하여 /user/activate/에 POST 요청을 보내 사용자를 활성화할 수 있습니다.

# src/resources.py
import hashlib
import http
import uuid

from flama import types
from flama.ddd.exceptions import NotFoundError
from flama.exceptions import HTTPException
from flama.http import APIResponse
from flama.resources import Resource, resource_method

from src import models, schemas, worker

__all__ = ["AdminResource", "UserResource"]

ENCRYPTION_SALT = uuid.uuid4().hex
ENCRYPTION_PEPER = uuid.uuid4().hex

class Password:
    def __init__(self, password: str):
        self._password = password

    def encrypt(self):
        return hashlib.sha512(
            (hashlib.sha512((self._password + ENCRYPTION_SALT).encode()).hexdigest() + ENCRYPTION_PEPER).encode()
        ).hexdigest()

class UserResource(Resource):
    name = "user"
    verbose_name = "User"

    @resource_method("/", methods=["POST"], name="create")
    async def create(self, worker: worker.RegisterWorker, data: types.Schema[schemas.UserDetails]):
        """
        tags:
            - User
        summary:
            User create
        description:
            Create a user
        responses:
            200:
                description:
                    User created in successfully.
        """
        async with worker:
            try:
                await worker.user.retrieve(email=data["email"])
            except NotFoundError:
                await worker.user.create({**data, "password": Password(data["password"]).encrypt(), "active": False})

        return APIResponse(status_code=http.HTTPStatus.OK)

    @resource_method("/signin/", methods=["POST"], name="signin")
    async def signin(self, worker: worker.RegisterWorker, data: types.Schema[schemas.UserCredentials]):
        """
        tags:
            - User
        summary:
            User sign in
        description:
            Create a user
        responses:
            200:
                description:
                    User signed in successfully.
            401:
                description:
                    User not active.
            404:
                description:
                    User not found.
        """
        async with worker:
            password = Password(data["password"])
            try:
                user = await worker.user.retrieve(email=data["email"])
            except NotFoundError:
                raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND)

            if user["password"] != password.encrypt():
                raise HTTPException(status_code=http.HTTPStatus.UNAUTHORIZED)

            if not user["active"]:
                raise HTTPException(
                    status_code=http.HTTPStatus.BAD_REQUEST, detail=f"User must be activated via /user/activate/"
                )

        return APIResponse(status_code=http.HTTPStatus.OK, schema=types.Schema[schemas.User], content=user)

    @resource_method("/activate/", methods=["POST"], name="activate")
    async def activate(self, worker: worker.RegisterWorker, data: types.Schema[schemas.UserCredentials]):
        """
        tags:
            - User
        summary:
            User activate
        description:
            Activate an existing user
        responses:
            200:
                description:
                    User activated successfully.
            401:
                description:
                    User activation failed due to invalid credentials.
            404:
                description:
                    User not found.
        """
        async with worker:
            try:
                user = await worker.user.retrieve(email=data["email"])
            except NotFoundError:
                raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND)

            if user["password"] != Password(data["password"]).encrypt():
                raise HTTPException(status_code=http.HTTPStatus.UNAUTHORIZED)

            if not user["active"]:
                await worker.user.update({**user, "active": True}, id=user["id"])

        return APIResponse(status_code=http.HTTPStatus.OK)

    @resource_method("/deactivate/", methods=["POST"], name="deactivate")
    async def deactivate(self, worker: worker.RegisterWorker, data: types.Schema[schemas.UserCredentials]):
        """
        tags:
            - User
        summary:
            User deactivate
        description:
            Deactivate an existing user
        responses:
            200:
                description:
                    User deactivated successfully.
            401:
                description:
                    User deactivation failed due to invalid credentials.
            404:
                description:
                    User not found.
        """
        async with worker:
            try:
                user = await worker.user.retrieve(email=data["email"])
            except NotFoundError:
                raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND)

            if user["password"] != Password(data["password"]).encrypt():
                raise HTTPException(status_code=http.HTTPStatus.UNAUTHORIZED)

            if user["active"]:
                await worker.user.update({**user, "active": False}, id=user["id"])

        return APIResponse(status_code=http.HTTPStatus.OK)

로그인 후 복사

이 요청을 통해 사용자가 활성화되어야 하며 본문이 비어 있는 200 응답을 받아야 합니다.

이전 사례와 마찬가지로 누군가가 잘못된 비밀번호로 사용자를 활성화하려고 하면 어떻게 되는지 테스트할 수도 있습니다.

poetry add "flama[full]" "aiosqlite"

로그인 후 복사

이 경우 다음 본문과 함께 401 응답을 받아야 합니다.

# src/app.py
from flama import Flama

app = Flama(
    title="Domain-driven API",
    version="1.0.0",
    description="Domain-driven design with Flama ?",
    docs="/docs/",
)

@app.get("/", name="info")
def info():
    """
    tags:
        - Info 
    summary:
        Ping
    description:
        Returns a brief description of the API
    responses:
        200:
            description:
                Successful ping.
    """
    return {"title": app.schema.title, "description": app.schema.description, "public": True}

로그인 후 복사

마지막으로 존재하지 않는 사용자를 활성화해야 합니다.

flama run --server-reload src.app:app

INFO:     Started server process [3267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

로그인 후 복사

이 경우 다음 본문과 함께 404 응답을 받아야 합니다.

# src/__main__.py
import flama

def main():
    flama.run(
      flama_app="src.app:app", 
      server_host="0.0.0.0", 
      server_port=8000, 
      server_reload=True
    )

if __name__ == "__main__":
    main()

로그인 후 복사

활성화 후 사용자 로그인

이제 사용자가 활성화되었으므로 다시 로그인을 시도할 수 있습니다.

poetry run python src/__main__.py

INFO:     Started server process [3267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

로그인 후 복사

이번에는 사용자 정보와 함께 200 응답을 반환해야 합니다.

# src/models.py
import uuid

import sqlalchemy
from flama.sqlalchemy import metadata
from sqlalchemy.dialects.postgresql import UUID

__all__ = ["user_table", "metadata"]

user_table = sqlalchemy.Table(
    "user",
    metadata,
    sqlalchemy.Column("id", UUID(as_uuid=True), primary_key=True, nullable=False, default=uuid.uuid4),
    sqlalchemy.Column("name", sqlalchemy.String, nullable=False),
    sqlalchemy.Column("surname", sqlalchemy.String, nullable=False),
    sqlalchemy.Column("email", sqlalchemy.String, nullable=False, unique=True),
    sqlalchemy.Column("password", sqlalchemy.String, nullable=False),
    sqlalchemy.Column("active", sqlalchemy.Boolean, nullable=False),
)

로그인 후 복사

사용자 비활성화

마지막으로 사용자의 자격 증명을 사용하여 /user/deactivate/에 POST 요청을 보내 사용자를 비활성화할 수 있습니다.

# migrations.py
from sqlalchemy import create_engine

from src.models import metadata

if __name__ == "__main__":
    # Set up the SQLite database
    engine = create_engine("sqlite:///models.db", echo=False)

    # Create the database tables
    metadata.create_all(engine)

    # Print a success message
    print("Database and User table created successfully.")

로그인 후 복사

이 요청을 사용하면 사용자가 비활성화되고 본문이 비어 있는 200 응답을 받아야 합니다.

결론

이 게시물에서 우리는 DDD(도메인 기반 디자인)의 세계와 이를 flama 애플리케이션에서 구현하는 방법을 살펴보았습니다. 우리는 DDD가 애플리케이션의 비즈니스 논리를 데이터 액세스 논리에서 분리하는 데 어떻게 도움이 되는지, 그리고 이러한 관심사의 분리가 어떻게 코드를 보다 유지 관리 및 테스트 가능하게 만드는지 살펴보았습니다. 또한 flama 애플리케이션에서 저장소와 작업자 패턴을 구현하는 방법과 이를 사용하여 데이터 액세스 논리를 캡슐화하고 수행해야 하는 모든 작업을 그룹화하는 방법을 제공하는 방법도 살펴보았습니다. 단일 트랜잭션 내의 데이터 소스에 대해. 마지막으로 리소스 메서드를 사용하여 사용자 데이터와 상호 작용하는 API 엔드포인트를 정의하는 방법과 DDD 패턴을 사용하여 이 예의 시작 부분에서 제공한 비즈니스 요구 사항을 구현하는 방법을 살펴보았습니다.

여기에서 설명한 로그인 프로세스가 완전히 현실적이지는 않지만, 이 자료와 JWT 인증에 대한 이전 게시물을 결합하여 보다 현실적인 프로세스를 구현할 수 있습니다. JWT 토큰. 이에 관심이 있으신 분들은 flama를 통한 JWT 인증 포스팅을 확인하실 수 있습니다.

이 게시물이 도움이 되기를 바라며, 이제 귀하의 flama 애플리케이션에 DDD를 구현할 준비가 되셨기를 바랍니다. 질문이나 의견이 있으시면 언제든지 저희에게 연락해 주세요. 우리는 언제나 기꺼이 도와드리겠습니다!

flama와 AI 및 소프트웨어 개발 세계의 기타 흥미로운 주제에 대한 더 많은 게시물을 기대해 주세요. 다음 시간까지!