ML API의 개발 및 생산화에 도움이 되는 몇 가지 흥미로운 새 기능을 제공하는 Flama 1.7의 최근 릴리스에 대해 이미 들어보셨을 것입니다. 이 게시물은 해당 릴리스의 주요 특징 중 하나인 도메인 기반 디자인 지원에 대해 자세히 설명합니다. 그러나 실제 사례를 통해 세부 사항을 살펴보기 전에 다음 리소스를 염두에 두는 것이 좋습니다(아직 숙지하지 않았다면 숙지하시기 바랍니다).
이제 새로운 기능을 시작하고 이를 활용하여 강력하고 유지 관리가 가능한 ML API를 구축하는 방법을 살펴보겠습니다.
이 게시물의 구성은 다음과 같습니다.
현대 소프트웨어 개발에서는 비즈니스 로직을 애플리케이션의 기술 설계에 맞추는 것이 필수적입니다. 이것이 도메인 중심 설계(DDD)가 빛을 발하는 곳입니다. DDD는 비즈니스의 핵심 영역을 반영하는 소프트웨어 구축을 강조하고, 비즈니스 개념을 중심으로 코드를 구성하여 복잡한 문제를 해결합니다. 이를 통해 DDD는 개발자가 유지 관리 가능하고 확장 가능하며 강력한 응용 프로그램을 만들 수 있도록 도와줍니다. 다음에서는 여러분이 알아야 할 DDD의 가장 중요한 개념을 소개합니다. 이에 대해 자세히 알아보기 전에 이 게시물은 DDD에 대한 포괄적인 가이드도 아니고 해당 주제에 대한 주요 참고 자료를 대체할 의도도 없다는 점을 말씀드리고 싶습니다. 실제로 DDD에 대해 더 깊이 이해하려면 다음 리소스를 권장합니다.
DDD의 주요 개념을 자세히 알아보기 전에 Cosmic Python의 매우 유용한 그림을 살펴보는 것이 좋습니다. 이 그림에서는 이러한 개념이 앱의 맥락에서 표시되어 서로 어떻게 연결되어 있는지 보여줍니다. .
도메인 모델의 개념은 간단한 용어 정의로 설명할 수 있습니다.
따라서 도메인 모델은 비즈니스 소유자가 비즈니스 운영 방식에 대해 염두에 두고 있는 일련의 개념과 규칙을 참조하는 멋진(그러나 표준적이고 유용한) 방법입니다. 이는 시스템의 동작을 제어하는 규칙, 제약 조건 및 관계를 포함하여 우리가 흔히 애플리케이션의 비즈니스 로직이라고 부르는 것입니다.
이제부터 도메인 모델을 모델이라고 부르겠습니다.
리포지토리 패턴은 모델과 데이터 액세스를 분리할 수 있는 디자인 패턴입니다. 리포지토리 패턴의 기본 아이디어는 데이터 액세스 논리와 애플리케이션의 비즈니스 논리 사이에 추상화 계층을 만드는 것입니다. 이 추상화 계층을 사용하면 문제를 분리하여 코드를 더욱 유지 관리하고 테스트할 수 있습니다.
저장소 패턴을 구현할 때 일반적으로 다른 저장소에서 구현해야 하는 표준 메서드를 지정하는 인터페이스(AbstractRepository)를 정의합니다. 그런 다음 데이터 액세스 논리가 구현되는 이러한 메서드의 구체적인 구현으로 특정 저장소가 정의됩니다(예: SQLAlchemyRepository). 이 디자인 패턴은 데이터 조작 방법을 분리하여 애플리케이션의 다른 곳에서 원활하게 사용할 수 있도록 하는 것을 목표로 합니다. 우리 도메인 모델에서.
작업 패턴 단위는 데이터 액세스에서 모델을 최종적으로 분리하는 데 누락된 부분입니다. 작업 단위는 데이터 액세스 논리를 캡슐화하고 단일 트랜잭션 내에서 데이터 소스에 대해 수행되어야 하는 모든 작업을 그룹화하는 방법을 제공합니다. 이 패턴을 사용하면 모든 작업이 원자적으로 수행됩니다.
작업 단위 패턴을 구현할 때 일반적으로 다른 작업 단위가 구현해야 하는 표준 메서드(AbstractUnitOfWork)를 지정하는 인터페이스를 정의합니다. 그런 다음 데이터 액세스 논리가 구현되는 이러한 메서드의 구체적인 구현을 통해 특정 작업 단위가 정의됩니다(예: SQLAlchemyUnitOfWork). 이 디자인을 사용하면 애플리케이션의 비즈니스 로직 구현을 변경할 필요 없이 데이터 소스에 대한 연결을 체계적으로 처리할 수 있습니다.
DDD의 주요 개념을 간략하게 소개한 후 Flama를 사용하여 DDD를 구현하는 방법을 알아볼 준비가 되었습니다. 이 섹션에서는 Flama를 사용하여 개발 환경을 설정하고, 기본 애플리케이션을 구축하고, DDD 개념을 구현하는 과정을 안내합니다.
예를 진행하기 전에 방금 검토한 주요 DDD 개념과 관련된 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 개념 구현을 시작할 수 있습니다.
실제 시나리오를 모방하려는 간단한 예의 맥락입니다. 사용자 관리를 위한 API 개발을 요청받았고, 다음과 같은 요구 사항이 제공되었다고 가정해 보겠습니다.
이 요구 사항 집합은 이전에 애플리케이션의 도메인 모델이라고 불렀던 것으로 구성되며, 이는 기본적으로 다음 사용자 워크플로를 구체화한 것에 지나지 않습니다.
이제 저장소와 작업자 패턴을 사용하여 도메인 모델을 구현해 보겠습니다. 먼저 데이터 모델을 정의한 다음 저장소와 작업자 패턴을 구현하겠습니다.
사용자 데이터는 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 작업을 수행하는 메서드 집합을 제공합니다. 특히 다음과 같습니다.
이 예에서는 테이블에 대한 추가 작업이 필요하지 않으므로 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)
이제 데이터 모델, 저장소 및 작업자 패턴, 리소스 메서드를 구현했으므로 이전에 소개한 기본 애플리케이션을 수정하여 모든 것이 예상대로 작동하도록 해야 합니다. 우리는 다음을 수행해야 합니다.
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)
flama에서 제공하는 자동 생성 문서 UI를 사용하여 이 작업을 시도할 수 있습니다. 거기에서 엔드포인트를 시도합니다.
# 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)
로그인
# 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
# 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
# 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
# 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)
이전 사례와 마찬가지로 누군가가 잘못된 비밀번호로 사용자를 활성화하려고 하면 어떻게 되는지 테스트할 수도 있습니다.
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 및 소프트웨어 개발 세계의 기타 흥미로운 주제에 대한 더 많은 게시물을 기대해 주세요. 다음 시간까지!
저희가 하는 일이 마음에 드신다면 저희 업무를 무료로 쉽게 지원할 수 있는 방법이 있습니다. Flama에서 ⭐를 선물해보세요.
GitHub ⭐는 우리에게 세상을 의미하며, 강력한 기계 학습 API를 구축하는 여정에서 다른 사람들을 돕기 위해 계속 노력할 수 있는 가장 좋은 원동력을 제공합니다.
AI, 소프트웨어 개발 등에 대한 흥미로운 스레드 외에도 최신 뉴스와 업데이트를 공유하는 ?에서 우리를 팔로우할 수도 있습니다.
위 내용은 Flama를 사용한 기본 도메인 기반 디자인의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
