안녕하세요. 휴먼스케이프 june입니다.

오늘은 Koa 기반의 프로젝트에 NestJS를 적용한 과정을 공유하려고 합니다.

거대해지는 서비스 레이어

우리가 참여한 서비스는 기존에 Koa로 개발되었고, MVC 패턴으로 설계되었습니다. 초기에는 간단한 서비스를 개발하기에 적합한 직관적이고 경제적인 구조였지만, 비즈니스 로직이 점점 복잡해지면서 코드의 가독성과 의존성 관리가 어려워지며 함수 재사용이 어려운 상황이었습니다.

우리는 코드 가독성을 향상시키고 관심사를 분리하기 위해 다양한 방법을 고려했습니다.

우리는 아래의 여러 장점을 이유로 NestJS를 도입하기로 결정했습니다.

1. 코드 가독성과 유지보수성 향상

• koa 기반 프로젝트는 구조의 자유도가 높아 팀마다 원하는 구조를 가져갈 수 있고, 우리의 Koa 기반 프로젝트는 단순하고 직관적인 구조로 개발되었습니다. 하지만 비즈니스 로직이 복잡해지고 코드베이스가 커지면서 현재의 구조로 확장성 있는 개발을 기대하기 어려워 졌습니다.
• NestJS는 모듈, 컨트롤러, 서비스 등의 구조를 제공하여 코드를 구성하고 관리할 수 있는 강력한 구조를 제공합니다. 이를 통해 코드를 모듈화하고 관심사를 분리하여 가독성을 향상시킬 수 있습니다.

2. 구조 강제와 표준화

• NestJS는 의존성 주입과 같은 특성을 활용하여 구조를 강제하고, 일관된 코드 작성을 장려합니다. 이는 팀원 간의 협업이 원활해지고, 코드의 일관성과 유지보수성을 향상시킵니다.

NestJS는 어댑터 패턴으로 작성되어 있으며, nest-koa-adapter 라이브러리를 사용하여 Koa와 NestJS를 혼용하면서 자연스럽게 NestJS로 전환할 수 있으리란 기대가 있었습니다.

Nestjs도입 외에 아래와 같은 옵션도 고려했습니다.

• Koa 프로젝트의 코드 구조를 개선하는 방법을 고민했습니다. 기존 로직을 수정하는 작업이기 때문에 러닝 커브가 적을 수 있었지만, 구조를 강제하기 어려워 다시 이전 구조로 돌아갈 가능성이 크다는 판단을 했습니다. 또, 기존 로직을 해석하고 수정하는 대신 코드를 새로 작성하는 것이 더 쉬울 수 있다고 판단했습니다.
• 각 서비스를 Microservice로 분리하는 방안도 고려했습니다(MSA). 그러나 각 서비스를 배치해야 할 인력 부족과 분리된 서비스 간 통신을 위한 API 작성에 드는 시간 대비 가성비가 좋지 않다고 판단했습니다.

Nestjs 도입기

위의 장단점을 고려한 후, 우리는 NestJS를 도입하기로 결정했습니다. 기존 Koa 프로젝트의 app.js에서 nest-koa-adapter를 사용하여 NestJS를 연결했습니다.

async function bootstrap() {
  const koa = new Koa();
  // error-handling
  koa.on('error', (err, ctx) => {
    console.error('onerror', err.stack);
    sentry.withScope((scope) => {
      scope.addEventProcessor(async (event) =>
        sentry.Handlers.parseRequest(event, ctx.request),
      );
      const { userId, email, deviceId } = ctx.req.user ?? {};
      scope.setUser({ userId, email, deviceId });
      sentry.captureException(err);
    });
  });
  koa.proxy = true  

  const app = await NestFactory.create<NestKoaApplication>(
    AppModule,
    new KoaAdapter(koa),
  );

  app.use(cors());
  app.use(helmet());

하지만 문제가 발생했습니다. NestJS와 Koa에서 사용하는 라이브러리 간 충돌이 발생하여 Nestjs와 Koa를혼용할 수 없는 상황이었습니다. 앞으로 발생할 수 있는 이슈를 고려하여 라이브러리 충돌을 해결하는 것보다는 독립된 환경으로 빠르게 전환하는 것이 좋다고 판단했습니다.

Docker-compose를 이용한 다중 컨테이너 오케스트레이션

따라서 우리는 Docker Compose를 사용하여 NestJS와 Koa 서버를 동시에 실행하고, Nginx의 리버스 프록시를 이용하여 /v1/ 경로는 NestJS로, 그 외의 경로는 Koa로 라우팅되도록 설정했습니다.

혹시 모를 충돌이 우려되어 npm monorepo를 사용하지 않고, 두 개의 프로젝트를 완전히 분리시켜 개발했습니다.

Docker-compse.yaml

version: '3.8'
services:
  koa:
    build:
      context: .
      dockerfile: ./Dockerfile.koa
  nest:
    extra_hosts:
          - 'host.docker.internal:host-gateway'
    build:
      context: .
      dockerfile: ./Dockerfile.nest
  nginx:
    container_name: nginx
    image: nginx
    restart: always
    ports:
      - "80:80"
    depends_on:
      - koa
      - nest
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf

nginx.conf

user  nginx;
worker_processes  auto;
error_log  /var/log/nginx/error.log warn;
pid        /var/run/nginx.pid;
events {
    worker_connections  1024;
}

http {
    upstream koa {
        server koa:3000;
    }

    upstream nest {
        server nest:4000;
    }

    server {
        listen 80;

        location / {
        proxy_pass http://koa/;
        proxy_http_version 1.1;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        }

        location /v1/ {
            proxy_pass http://nest/;
            proxy_http_version 1.1;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        }
    }
}

회고

위의 작업을 수행하며 느낀 아쉬운점과 잘한점은 다음과 같습니다.

아쉬운점

• 낯선 프레임워크를 도입하려다 보니 코드 한줄 한줄에 대해 고민하는 시간이 길어졌습니다. 때로는 고민보다 빠르게 개발해보고 다시 수정하는게 나을 수 있다는 생각을 하게 되었습니다.

잘한점

• nestjs 프로젝트 개발 중 이슈가 발생했을 때 이슈의 원인을 정확하게 분석하고, 라이브러리 충돌이라는 결론이 나왔을 때 빠르게 Docker-compose를 이용하는 쪽으로 의사결정하는 과정이 결과적으로 빠르게 nestjs를 도입하는데에 큰 도움이 되었습니다.

NestJS와 Koa 서비스를 동시에 실행한 후에는 점차적으로 NestJS에서 비즈니스 로직을 재작성하고 새로운 구조에 대해 지속적으로 고민하고 있습니다. 컨버팅 작업이 완료된 후에는 컨버팅 후기를 다시 공유하도록 하겠습니다.

이상으로 Koa 프로젝트에 NestJS를 적용한 과정과 레거시 개선을 통한 코드 구조 개선에 대해 공유드렸습니다.

Koa project NestJS 적용기: 레거시 개선을 통한 코드 구조 개선 was originally published in None on Medium, where people are continuing the conversation by highlighting and responding to this story.

안녕하세요. 휴먼스케이프 june입니다.

이번에는 개발 중인 서비스에 도커를 도입하면서 겪은 경험에 대해 공유하고자 합니다.

이전 글에서 언급한 대로, 우리는 임상 연구 데이터를 적재하는 크론 서버를 운영하고 있었습니다. 그러나 시간이 흐를수록 EC2 환경이 조금씩 변경되어 배포에 실패하는 문제가 발생했습니다. 이 문제를 해결하면서 아래의 추가적인 이점들을 가져가기 위해 도커를 도입하기로 결정했습니다.

1. 일관된 환경: Docker는 이미지를 통해 개발 환경과 프로덕션 환경 간의 일관성을 제공합니다. 개발 환경에서 이미지를 빌드하고, 해당 이미지를 프로덕션 환경에서 실행함으로써 일관성을 유지할 수 있습니다.
2. 빠른 배포: Docker는 이미지를 사용하여 애플리케이션을 빠르게 배포할 수 있습니다. 이미지는 가볍고 빠르게 생성되며, 필요한 경우 이미지를 공유하고 재사용할 수 있습니다.
3. 확장성: Docker는 컨테이너 단위로 애플리케이션을 실행하므로, 필요에 따라 애플리케이션의 인스턴스 수를 쉽게 확장할 수 있습니다. 이는 애플리케이션의 성능과 가용성을 향상시키는 데 도움이 됩니다.
4. 격리와 보안: Docker는 각 컨테이너를 격리된 환경에서 실행하므로, 애플리케이션 간의 상호 영향을 최소화하고 보안을 강화할 수 있습니다.

Docker 도입

도커를 도입하기 전에, 도커의 장점인 일관적인 환경 제공과 빠른 배포를 활용하려면 도커의 작동 방식에 대한 이해가 필요합니다. 도커는 Dockerfile의 명령문을 순차적으로 읽어가며 환경을 구성합니다. 이전에 동일한 Dockerfile을 사용하여 환경을 구성한 적이 있다면, 레이어 캐시를 사용하여 명령문 단위로 이전 구성을 가져와 적용합니다. 또한 Dockerfile의 첫 줄에서 FROM 키워드를 사용하여 이미 빌드된 이미지를 다운로드하여 사용할 수 있습니다.

이러한 이해를 바탕으로 빠른 배포를 위한 방법을 고민하였습니다.

우리는 세 가지 레포지토리 구조를 고려했습니다.

1. 레포지토리를 사용하지 않고 배포 시마다 Beanstalk 또는 EC2에서 이미지를 빌드하여 사용하는 방법.
2. 빌드 시간이 오래 걸리고 자주 변경되지 않는 부분(종속성 패키지 설치 등)을 base 도커 파일로 분리하고, base docker image를 다운로드하여 소스 코드만 복사하여 사용하는 방법.
3. 각 환경마다 별도의 레포지토리(api, cron)를 만들어 배포 시마다 빌드하여 사용하는 방법.

위의 방법 중 우리는 2번 방법을 택했습니다.

각 방법의 장단점은 다음과 같이 정리했습니다.

1. 레포지토리를 사용하지 않아 요금은 증가하지 않지만, 매번 API/Cron 서버에서 빌드를 해야 하므로 원하는 빠른 배포를 실현하기 어려워 보였습니다.
2. 하나의 레포지토리만 사용하고, 설치할 패키지 변경 사항이 없으면 이미지를 업로드하지 않아 배포 시간을 단축할 수 있는 효과가 있었습니다. 또한 API/Cron 서버에서 소스 코드를 복사하고 프로그램을 실행하는 시간도 빠르므로 효율적인 방식으로 판단했습니다.
3. 여러 개의 레포지토리를 사용하여 비용이 증가하고, GitHub Actions에서 여러 번 업로드하는 문제로 인해 빌드 시간이 늘어날 것으로 예상되었습니다.

먼저 Dockerfile을 베이스, API, Cron으로 나누었습니다. 리눅스와 파이썬 패키지를 설치하고, 모든 환경에 공통적으로 적용되는 명령문은 base file에 넣었습니다. API와 Cron 파일에서는 각각 API와 Cron 서버에 배포하기 위해 소스 코드를 복사하고 프로그램을 실행하는 명령문을 추가했습니다.

Dockerfile.base

FROM python:3.8

ENV LANG C.UTF-8
ENV LC_ALL C.UTF-8
ENV LC_NUMERIC C.UTF-8
ENV LC_TIME C.UTF-8
ENV LC_COLLATE C.UTF-8
ENV LC_MONETARY C.UTF-8
ENV LC_MESSAGES C.UTF-8    
ENV PYTHONDONTWRITEBYTECODE 1
ENV TZ=Asia/Seoul

WORKDIR /app

COPY requirements.txt /app

RUN apt-get update
RUN apt-get install gcc g++ libffi-dev build-essential nginx supervisor make gunicorn curl libssl-dev libbz2-dev libffi-dev cron git libsqlite3-dev  -y
RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone

RUN pip3 install -r requirements.txt

Dockerfile.api

FROM ecr_image

COPY . /app
WORKDIR /app

COPY infra/docker-entrypoint.production.sh /docker-entrypoint.production.sh

EXPOSE 80

RUN chmod +x /docker-entrypoint.production.sh

Github action을 이용한 docker image build

우리는 GitHub Actions에서 베이스 이미지를 빌드하고 ECR로 업로드하는 과정에서 레이어 캐시를 사용하여 빌드 시간을 줄이려고 노력했습니다. build-push-action을 사용하여 buildx를 이용해 도커 이미지를 빌드하며, 이 과정에서 GitHub Actions 캐시를 사용하여 레이어 캐시를 활용했습니다.

그러나 해당 스크립트에 문제가 있었는데, buildx toolkit을 사용하다 보니 멀티 플랫폼을 지원하는 이미지로 빌드되어 ECR에 정상적이지 않은 형태로 업로드되었습니다.

이 문제를 해결하기 위해 직접 빌드 스크립트를 작성했습니다.

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v1
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: ap-northeast-2

      - name: Login to Amazon ECR
        id: login-ecr
        uses: aws-actions/amazon-ecr-login@v1

      - name: pull base image
        continue-on-error: true
        env:
          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
          ECR_REPOSITORY: repository_name
        run: |
          docker pull ${{ env.ECR_REGISTRY }}/${{ env.ECR_REPOSITORY }}:latest

      - name: push base image
        env:
          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
          ECR_REPOSITORY: repository_name
        run: |
          docker build --cache-from=${{ env.ECR_REGISTRY }}/${{ env.ECR_REPOSITORY }}:latest -f ./infra/Dockerfile.base -t tag_name .
          docker tag image_name ${{ env.ECR_REGISTRY }}/${{ env.ECR_REPOSITORY }}:latest
          docker tag image_name ${{ env.ECR_REGISTRY }}/${{ env.ECR_REPOSITORY }}:${{ github.sha }}
          docker push ${{ env.ECR_REGISTRY }}/${{ env.ECR_REPOSITORY }} --all-tags

회고

위의 작업들을 진행하며 아쉬운점, 잘한점이 생겼습니다.

아쉬운점은 다음과 같습니다.

• 작성한 빌드 스크립트에서는 API와 Cron 서버에서 도커 빌드 시 가장 최근의(latest) 태그가 붙은 이미지를 가져와 빌드하도록 되어 있었습니다. 하지만 롤백 등의 상황에서 최신 베이스 이미지를 가져와 빌드하면 문제가 발생할 수 있어 해당 부분을 개선해야 한다고 생각합니다.
• 위의 레포지토리 구조 고민에서 1번 방법을 선택하고 API/Cron 서버에서 레이어 캐시를 사용해 빌드하는 방식으로 진행했다면 속도 측면에서 큰 차이가 없었을 것으로 예상되는데, 이 부분을 인지하지 못하고 넘어간 것이 아쉬웠습니다.

잘한점은 다음과 같습니다.

• docker기반 배포의 이점인 빠른 배포 속도를 가져가기 위해 github actions에서 많은 삽질을 했고, 우리 상황에 맞는 방법을 찾아낸 것이 잘한점이라 생각됩니다.

지금까지 Docker 적용 과정에 대해 적어 보았습니다. 감사합니다!

Docker 적용기(aws ecr, github action) was originally published in None on Medium, where people are continuing the conversation by highlighting and responding to this story.

안녕하세요. 휴먼스케이프 june입니다.

오늘은 Django 기반의 서비스에서 크론 서버를 구성하는 과정에 대해 공유하려고 합니다. 이 글에서는 크론 서버 구성을 위해 우리가 겪었던 어려움과 선택한 해결책에 대해 설명하려 합니다.

크론 서버 구성

우리는 임상 연구를 적재하는 서비스를 개발하고 있었습니다. 처음에는 크론 서버 없이 필요할 때마다 API를 호출하여 임상 연구를 적재하는 방식으로 작업을 수행했습니다. 그러나 추가 리소스를 투입하지 않고도 주기적으로 임상 연구를 적재하기 위해 크론 서비스를 도입하게 되었습니다.

우리는 크게 2가지의 옵션을 두고 고려했습니다.

1. aws에서 제공하는 크론 관련 서비스 사용
2. Django에서 크론 작업을 관리하기 위해 django-crontab을 사용하고, EC2를 사용하여 관리

(Api 서버에 크론을 추가하지 않고 크론서버를 따로 만드는 이유는 이후 auto scaling시 같은 크론 작업이 두번 돌아갈 수 있기 때문입니다.)

첫 번째 옵션인 AWS의 크론 관련 서비스는 AWS 콘솔을 통해 동작 상황을 쉽게 파악할 수 있으며, AWS에서 제공하는 다양한 옵션을 통해 확장성을 기대할 수 있었습니다. 하지만 관리 포인트가 증가하고 코드로 크론을 관리할 수 없는 단점이 있었습니다.

두 번째 옵션인 Django와 django-crontab을 사용하여 EC2에 크론 작업을 관리하는 방법은 구조가 직관적이고 코드로 크론을 관리할 수 있는 장점이 있었습니다.

우리는 고민 끝에 두 번째 옵션을 선택하여 Django 서비스를 EC2에 배포하고, django-crontab 패키지를 사용하여 크론 작업을 관리하기로 결정했습니다.

django-crontab을 통한 cronjob 관리

크론은 크게 크론 데몬과 크론 테이블로 구성됩니다. 크론 데몬은 운영체제에서 동작하는 백그라운드 프로세스로, 정해진 시간에 크론 테이블에 등록된 작업을 실행합니다. 크론 테이블은 크론 작업의 목록을 관리하는 데이터베이스나 텍스트 파일로, 각 작업의 실행 주기와 명령어 등을 포함합니다.

django-crontab 패키지를 사용하면 Django 애플리케이션 내에서 크론 작업을 등록할 수 있습니다. 먼저, settings.py 파일에서 크론 작업을 등록할 수 있는 크론 설정(CRONJOBS)을 추가해야 합니다. 이 설정은 크론 작업의 주기와 실행할 함수 또는 커맨드를 지정합니다.

예를 들어, 매일 오전 3시에 실행되는 크론 작업을 등록하고 싶다면, settings.py 파일에 다음과 같은 내용을 추가합니다

CRONJOBS = [
    ('0 3 * * *', 'myapp.cron.my_cron_job')
]

위의 예시에서 '0 3 * * *'는 크론 표현식으로, 매일 오전 3시에 실행됨을 나타냅니다. 'myapp.cron.my_cron_job'은 실행할 함수의 경로를 나타내며, 이는 실제로 실행될 크론 작업 함수를 가리킵니다.

우리는 테스트의 용이성을 위해 작업들을 django command로 만든 후, command를 호출하는 함수를 CRONJOBS로 등록했습니다.

django-crontab 패키지를 사용하면 Django 애플리케이션과 크론 작업을 함께 버전 관리할 수 있습니다. 크론 작업을 Django 프로젝트 내에서 관리하면 코드베이스의 일부로서 크론 작업을 유지하고, 배포 및 관리가 편리해집니다. 또한, Django의 ORM과 다른 기능들을 크론 작업 내에서 활용할 수 있어 개발과 유지보수가 용이해집니다.

이와 같은 방식으로 Django 애플리케이션과 크론 작업을 통합하면, 크론 작업을 실행해야 할 시간과 주기를 Django 설정 파일에서 관리할 수 있으며, Django 애플리케이션의 다른 부분과 일관된 방식으로 크론 작업을 개발하고 관리할 수 있습니다.

Slack Webhook 도입

우리는 크론 서버의 상태를 실시간으로 모니터링하고, 잠재적인 문제를 빠르게 감지하기 위해 Slack webhook을 도입했습니다.

webhook은 크론 작업이 실행될 때마다 성공 여부를 확인하는 메시지를 전송하는 방식으로 개발하였습니다.

Slack의 경우 API 제공이 원활하고, 사내 메신저로 사용중이기 때문에 접근성이 좋아 사용하기로 했습니다.

예기치 않은 상황으로 인 크론 서버가 다운되는 경우, Slack webhook을 통한 이슈 전달이 불가능할 수도 있습니다. 이를 고려하여 우리는 크론 작업이 실행될 때마다 성공 여부를 확인하는 메시지를 전송하는 방식으로 개발하였습니다.

이를 통해 우리의 시스템은 안정성을 유지하며, 개발자들은 크론 서버의 동작 상태에 대해 쉽고 편리하게 알 수 있게 되었습니다.

크론서버 docker 도입

그러나 문제가 발생했습니다. 배포할 때마다 EC2 환경이 조금씩 달라져서 배포에 실패하는 경우가 발생했습니다. 이 문제를 해결하기 위해 우리는 Docker를 도입하기로 결정했습니다. Docker를 사용하여 항상 동일한 서버 환경을 제공하여 환경 구성과 관련된 문제 및 변수를 제거하고자 했습니다.

우리는 Docker를 사용할 수 있는 여러 서비스를 조사했습니다. 크게 네 가지 옵션이 있었습니다.

1. 기존 EC2에 Docker를 설치하여 사용하는 방법
2. AWS Batch 사용
3. Beanstalk Worker 사용
4. ECS 사용

우리는 인프라 의존성을 줄이고자 코드로 가능한 많은 부분을 관리하려고 했으며, 실제로 동작하는 크론 작업은 Django admin을 통해 모두 파악할 수 있었기 때문에 AWS 콘솔을 통해 현황을 파악하는 일이 거의 없었습니다. 따라서 우리는 해야 할 작업이 적고 러닝 커브가 낮은 EC2를 선택했습니다. 도커 이미지는 AWS ECR을 통해 관리하기로 결정했습니다.

Docker를 사용할 수 있는 여러 서비스를 조사하며 알아낸 장단점은 다음과 같습니다.

• EC2에 Docker를 설치하여 사용하는 방법은 EC2를 재생성해야 할 경우 Codedeploy 서비스와 연결하고, Docker를 설치하기 위해 SSH로 서버에 접근하여 작업해야 하는 단점이 있었습니다. 그러나 자유도가 높고 이전에 EC2를 사용하고 있었기 때문에 작업량이 적은 장점이 있었습니다.
• AWS Batch는 작업 당 작업 정의를 통해 세밀한 옵션 조정이 가능하며, 작업 현황을 쉽게 파악할 수 있는 장점이 있었습니다. 그러나 세밀한 옵션 조정이 필요하지 않은 경우 오히려 단점이 될 수 있었고, 우리는 이미 각 데이터의 처리 상태를 관리하기 위해 control_status_type이라는 컬럼을 사용하고 있었기 때문에 작업 현황 파악 기능이 필요하지 않았습니다.
• Beanstalk Worker는 EC2와 Batch 사이에 위치한 서비스로 보입니다. 세밀한 옵션 조정이 가능하지만 원하지 않는 경우에는 기본 옵션으로 작업을 진행할 수 있었으며, 이미 API 서버를 Beanstalk를 통해 서비스하고 있어 한꺼번에 관리하기 편리했습니다. 그러나 Beanstalk Worker를 사용하기 위해서는 기존에 커맨드로 만들어 둔 크론 작업을 API로 변환해야 하는 작업이 필요했습니다.
• ECS는 AWS Batch와 유사한 방식으로 사용되는 것으로 보였습니다.

회고

위에서 서술한 여러 작업을 진행한 후 스스로에 대한 회고를 간단하게 진행해 보았습니다.

아쉬운점은 다음과 같습니다.

• 위에서 서술한 ec2의 단점을 상쇄할만한 방법을 찾지 않고 작업을 마무리 한 점이 아쉬웠습니다. 현재는 ec2 인스턴스 개설 후 작성해야 하는 명령어 모음을 메모해두고 복붙하는 형태로 사용중입니다.
• slack webhook이 주기적으로 계속 날아오기 때문에 매번 확인해야 하는 작업자 입장에서 피로감이 생기게 됩니다. slack webhook이 정해진 시간내에 오지 않을 경우에만 알려주는 다른 서비스를 접목시켰으면 어땠을까 하는 생각이 있습니다.

잘한점은 다음과 같습니다.

• 인프라 의존성을 줄이고자 시도했고, 스타트업의 경우 크레딧을 얻어 클라우드 플랫폼을 옮기는 경우가 종종 있어 이 경우에 매우 유용했습니다.
• 남들이 많이 사용하는 서비스를 이용해 그냥 개발하기 보다 현재 상황을 정확히 파악하고, 상황에 맞는 구현방법을 찾아 오버/언더 엔지니어링을 피하고자 했습니다.

지금까지 Django 기반의 크론 서버 환경 구성에 대해 알아보았습니다. 다음 글에서는 도커 이미지 관리에 대해 자세히 다루도록 하겠습니다. 감사합니다!

cron서버 환경 구성 삽질 일지 was originally published in None on Medium, where people are continuing the conversation by highlighting and responding to this story.

안녕하세요. 휴먼스케이프 june입니다.

이번 시간엔 App을 추가하는 방법에 대해 알아보도록 하겠습니다.

프로젝트를 진행한 코드는 깃허브에 올라가 있으니 참고하시기 바랍니다.

GitHub - Root-kjh/drf_practice

App

app은 Django 프로젝트 내에서 특정 기능 또는 업무 영역을 담당하는 독립적인 모듈입니다. 각 앱은 자체적으로 데이터 모델, 뷰, 템플릿, URL 매핑, 정적 파일 등을 가질 수 있으며, Django 프로젝트 내에서 재사용이 가능합니다.

App을 구분하는 기준은 상황에따라, 사람마다 다를 수 있습니다.

저는 DDD(Domain-driven Design)의 Bounded Context 개념을 참고하여 app을 구분합니다.

방대한 프로젝트가 하나의 app에 들어가 있을 경우 복잡도가 급격하게 증가해 유지보수가 힘들어질 수 있고, 너무 잘게 쪼개면 각 app간의 의존성이 너무 많아져 관리하기 힘들어 질 수 있습니다.

따라서 상황에따라, 필요에따라 적절히 app을 나누어야 합니다.

App을 생성하는 방법은 다음과 같습니다.

python3 manage.py startapp users

위 명령어를 입력하면 users app이 생성됩니다.

settings의 INSTALLED_APPS에 users를 추가하면 해당 app을 사용할 수 있습니다.

개인적으로는 외부에서 가져온 app과 내부에서 작성한 app을 구분하기 위해 THIRD_PARTY_APPS, LOCAL_APPS 리스트를 나눠 작성하고 INSTALLED_APPS = THIRD_PARTY_APPS + LOCAL_APPS 로 합쳐서 사용합니다.

지금까지 배운 것들을 활용해 우리는 원하는 api들을 자유자재로 만들 수 있습니다.

하루만에 완성하는 Django+DRF 서비스는 여기서 마무리 짓도록 하겠습니다.

하루만에 완성하는 Django+DRF 서비스(5) — END was originally published in None on Medium, where people are continuing the conversation by highlighting and responding to this story.

안녕하세요. 휴먼스케이프 june입니다.

이번 시간엔 쿼리성능 개선 방법에 대해 알아보도록 하겠습니다.

프로젝트를 진행한 코드는 깃허브에 올라가 있으니 참고하시기 바랍니다.

GitHub - Root-kjh/drf_practice

쿼리 확인

django+drf의 쿼리를 확인할 수 있는 방법은 여러가지가 있습니다.

그 중 가장 간단한 방식 2가지를 소개드리도록 하겠습니다.

Django Debug Toolbar

GitHub - jazzband/django-debug-toolbar: A configurable set of panels that display various debug information about the current request/response.

문서를 보고 debug toolbar를 설치합니다.

서버 내의 모든 view에 툴바가 생긴것을 확인할 수 있습니다.

SQL탭을 누르면 어떤 쿼리가 실행되었는지, 쿼리 explain등을 볼 수 있습니다.

Query logging

django의 logging 문서를 참고해 settings.py에 로깅 설정을 작성합니다.

Django

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'handlers': {
        'console': {
            'level': 'DEBUG',
            'class': 'logging.StreamHandler',
        },
    },
    'loggers': {
        'django.db.backends': {
            'handlers': ['console'],
            'level': 'DEBUG',
        },
    },
}

api를 호출하면 console에 쿼리가 찍히는것을 확인할 수 있습니다.

(0.000) SELECT "django_session"."session_key", "django_session"."session_data", "django_session"."expire_date" FROM "django_session" WHERE ("django_session"."expire_date" > '2023-03-07 12:56:17.442700' AND "django_session"."session_key" = '2pa7rg8hnhtummtoem4pcr8nzb26mf1r') LIMIT 21; args=('2023-03-07 12:56:17.442700', '2pa7rg8hnhtummtoem4pcr8nzb26mf1r'); alias=default
(0.000) SELECT "auth_user"."id", "auth_user"."password", "auth_user"."last_login", "auth_user"."is_superuser", "auth_user"."username", "auth_user"."first_name", "auth_user"."last_name", "auth_user"."email", "auth_user"."is_staff", "auth_user"."is_active", "auth_user"."date_joined" FROM "auth_user" WHERE "auth_user"."id" = 1 LIMIT 21; args=(1,); alias=default
[07/Mar/2023 12:56:17] "GET /admin/jsi18n/ HTTP/1.1" 200 3343

쿼리 성능 개선

비효율적인 쿼리를 발생시키기 위해 item list view에 ItemDetailSerializer를 연결시켜보겠습니다.

class ItemViewSet(viewsets.ModelViewSet):
    queryset = Item.objects.all()
    serializer_class = ItemDetailSerializer
    permission_classes = [DjangoModelPermissionsOrAnonReadOnly]

    # def get_serializer_class(self):
    #     if self.action == 'retrieve':
    #         return ItemDetailSerializer
    #     return ItemSerializer

item list를 조회하게 되면 아래와 같이 각 item당 review를 검색하는 쿼리가 실행됩니다.

이렇게 object하나 당 해당 object에 연결된 objects를 가져오기 위해 쿼리를 추가로 실행하는 문제를 n+1문제라 부릅니다.

n+1문제를 해결하는 방법은 2가지가 있습니다.

selected_related

내부적으로 join을 사용해 데이터를 한번에 가져옵니다.

따라서 join으로 가져올 수 있는 관계 (연결된 object가 하나일경우)에서 사용됩니다.

prefetch_related

가져오는 objects의 id list를 이용해 in 연산으로 연결된 objects를 가져옵니다.

select_related와 달리 쿼리를 2개로 나누고, python에서 join합니다.

이번 경우엔 prefetch_related를 통해 쿼리를 최적화합니다.

Index

index는 데이터 검색 속도를 향상시키기 위해 사용하는 데이터 구조입니다. 인덱스는 특정 column 값을 색인화하여 데이터베이스에서 빠르게 검색할 수 있도록 합니다.

일반적으로 데이터를 검색할 때, 전체 데이터를 하나씩 탐색하면서 조건에 맞는 데이터를 찾습니다. 이는 데이터의 양이 적을 때는 큰 문제가 되지 않지만, 데이터의 양이 많아질수록 검색 속도가 느려지는 문제가 발생할 수 있습니다.

이 때, 인덱스는 데이터베이스에서 데이터를 빠르게 검색할 수 있도록 도와줍니다.

예를 들어, 특정 테이블에서 ‘이름’ 필드를 인덱스로 설정하면, 이름으로 검색하는 작업에서 데이터베이스는 전체 데이터를 탐색하는 것이 아니라, 인덱스를 사용하여 더 빠르게 해당 데이터를 찾아낼 수 있습니다.

인덱스는 데이터베이스에서 자주 검색되는 필드에 대해서 생성하는 것이 좋습니다.

데이터베이스에서 데이터를 검색할 때 빠르게 찾을 수 있는 장점이 있지만, 데이터를 추가, 수정, 삭제할 때 추가적인 작업이 필요하며, 인덱스의 크기가 크면 디스크 공간도 많이 차지하게 됩니다.

따라서 인덱스를 사용할 때에는 신중하게 결정해야 하며, 자주 사용하는 필드에 대해서만 생성하는 것이 좋습니다.

Django에서는 모델 클래스의 필드에 db_index=True 속성을 설정하여 해당 필드를 인덱스로 사용할 수 있습니다.

item의 date range로 검색하는 필터 기능이 추가된다고 가정해봅시다.

우리는 date 검색을 빠르게 하기 위해 date 필드에 db_index=True 속성을 설정할 수 있습니다.

class Item(models.Model):
    name = models.CharField(max_length=100)
    category = models.ForeignKey(Category, on_delete=models.CASCADE, related_name='items')
    description = models.TextField(null=True, blank=True)
    image = models.ImageField(upload_to='images/', null=True)
    date = models.DateTimeField(auto_now_add=True, db_index=True)

    def __str__(self):
        return self.name

지금까지 쿼리 최적화 방법에 대해 알아봤습니다.

다음 시간엔 새로운 앱을 추가해 여러 앱을 운용하는 방법에 대해 알아보도록 하겠습니다.

하루만에 완성하는 Django+DRF 서비스(4) was originally published in None on Medium, where people are continuing the conversation by highlighting and responding to this story.

안녕하세요. 휴먼스케이프 june입니다.

이번 시리즈에서는 해킹, 취약점에 대한 얘기를 자유형식으로 연재할 예정입니다.

그리고 이번 시간엔 Django 4.0.5버전에서 발견된 SQL Injection 취약점에 대해 얘기해보도록 하겠습니다.

사용하는 패키지관련 취약점 정보는 snyk사이트를 이용해 간편하게 확인할 수 있습니다.

django vulnerabilities | Snyk

POC는 github에 잘 올라와 있으며, 이번 편에서는 해당 POC를 기반으로 설명합니다.

GitHub - aeyesec/CVE-2022-34265: PoC for CVE-2022-34265 (Django)

SQL Injection

sql injection은 user의 입력값이 들어가는 곳에 sql을 넣어 비정상적인 동작을 하도록 유도하는것을 말합니다.

로그인 sql구문을 예로 들어보겠습니다.

select * from users where id=’{user_id}’ and password=’{password}’;

만약 user_id와 password에 ‘ or ‘’=’ 구문을 넣으면 모든 user를 가져오게 되며 로그인에 성공하게 됩니다.

select * from users where id=’' or ''='’ and password=’' or ''='’;

id가 test인 유저가 있을 경우 로그인에 통과하도록 할 수도 있습니다.

select * from users where id=’test’ and password=’' or ''='’;

이를 응용하면 db의 모든 정보들을 가져올 수 있습니다.

SQL Injection의 방어방법은 크게 2가지가 있습니다.

입력값 검증

• 입력값을 검증해 SQL Injection을 일으키는 입력값을 사전에 치환/제거할 수 있습니다.
• aws, cloudflare등에서 waf 등의 방화벽을 이용할 수 있습니다.

Prepared Statement 사용

• SQL Injection에 대한 방어방법으로 가장 권장되는 방식입니다.
• 쿼리파싱을 한 후 입력값을 바인딩합니다.
• 입력값은 단순 텍스트로 인식되어 쿼리에 전혀 영향을 끼치지 못합니다.

CVE-2022–34265

이번 취약점은 django orm의 Extract, Trunc함수에서 발견되었습니다.

Extract는 인자로 들어온 날짜의 단위를 숫자로 추출하는 함수, Trunc는 소수점을 버리듯이 인자로 들어온 단위(hour, day, month 등)까지 날짜를 잘라 반환하는 함수입니다.

날짜의 단위를 인자로 받는 부분에서 취약점이 발생하게 되는데, 해당 함수를 사용했을때의 실제 sql 구문을 보면 다음과 같습니다.

Extract

SELECT django_datetime_extract("year","start_datetime","UTC") AS "start_year"

여기서year 부분을 year’ FROM start_datetime)) OR ``=``;SELECT PG_SLEEP(5) --으로 바꾸게 되면 sql injection이 발생합니다.

(poc의 구문을 그대로 가져오면 해킹으로 인식되어 글 작성이 안되기 때문에 일부 수정했습니다.)

SELECT django_datetime_extract("year from table_name)) or ``=``;SELECT PG_SLEEP(5) --,"start_datetime",'UTC') AS "start_year"

POC에선 SELECT PG_SLEEP(5) 를 넣었지만, 데이터베이스 삭제 등의 위험한 쿼리도 충분히 실행할 수 있습니다.

Trunc

trunc도 extract와 거의 동일하게 동작합니다.

SELECT django_datetime_trunc('year', "start_datetime", 'UTC') AS "start_year"

year 부분을year’, start_datetime)) OR ``=``;SELECT PG_SLEEP(5)-- 으로 바꿔줍니다.

SELECT django_datetime_trunc('year', start_datetime)) OR ``=``;SELECT PG_SLEEP(5)--', "start_datetime", 'UTC') AS "start_year"

위와 같이 sql injection이 발생합니다.

해당 취약점들은 trunc,extract함수의 단위 인자를 api 파라미터 등의 유저 입력으로 받아오는 경우 발생합니다.

따라서 해당 인자들을 유저 입력으로 받아오지 않도록 수정하거나 django 버전을 4.0.6 이상으로 업데이트함을 권고드립니다.

개발자의 해킹 얘기 — N-day 취약점 리뷰(Django 4.0.5/SQL Injection) was originally published in None on Medium, where people are continuing the conversation by highlighting and responding to this story.

안녕하세요. 휴먼스케이프 june입니다.

이번 시간엔category, item, review의 모델 연결관계를 api에서 표현하는 방법, permission을 적용하는 방법(일반유저가 items를 삭제/수정하지 못하도록 관리)에 대해 알아보도록 하겠습니다.

프로젝트를 진행한 코드는 깃허브에 올라가 있으니 참고하시기 바랍니다.

GitHub - Root-kjh/drf_practice

이전 시간에 만든 category api, testcode와 같은 형식으로 item, review에 대한 api, testcode도 생성합니다.

자세한 코드는 github repo에서 확인해주세요.

from test_plus.test import APITestCase

from ..models import Item, Category

class ItemApiTestCase(APITestCase):
    def setUp(self) -> None:
        self.category = Category.objects.create(name='test category')
        self.item = Item.objects.create(name='test item', category=self.category, description='test description')
        self.item2 = Item.objects.create(name='test item 2', category=self.category, description='test description 2')

    def test_get_items(self):
        response = self.get('/reviews/items/')
        self.assert_http_200_ok(response)
        self.assertEqual(len(response.data), 2)
        self.assertEqual(response.data[0]['name'], self.item.name)
        self.assertEqual(response.data[1]['name'], self.item2.name)

    def test_get_item(self):
        response = self.get(f'/reviews/items/{self.item.id}/')
        self.assert_http_200_ok(response)
        self.assertEqual(response.data['name'], self.item.name)

    def test_create_item(self):
        data = {'name': 'test item 3', 'category': self.category.id, 'description': 'test description 3'}
        response = self.post('/reviews/items/', data=data)
        self.assert_http_201_created(response)
        self.assertEqual(response.data['name'], data['name'])
        self.assertEqual(Item.objects.count(), 3)
        self.assertEqual(Item.objects.last().name, data['name'])

    def test_update_item(self):
        data = {'name': 'test item 3', 'category': self.category.id, 'description': 'test description 3'}
        response = self.put(f'/reviews/items/{self.item.id}/', data=data)
        self.assert_http_200_ok(response)
        self.assertEqual(response.data['name'], data['name'])
        self.assertTrue(Item.objects.filter(name=data['name']).exists())
        self.assertFalse(Item.objects.filter(name=self.item.name).exists())

    def test_delete_item(self):
        response = self.delete(f'/reviews/items/{self.item.id}/')
        self.assert_http_204_no_content(response)
        self.assertFalse(Item.objects.filter(name=self.item.name).exists())
        self.assertEqual(Item.objects.count(), 1)

만든 api들을 확인해보면 잘 동작하는걸 알 수 있습니다.

Model의 연결관계 표현

우리는 category retrieve api를 호출할때 items를, item retrieve api를 호출할 때 reviews를 반환하도록 하고 싶습니다.

(클라이언트 단에서 api 호출을 적게 할수록 로딩 시간이 줄어듭니다.)

CategoryDetailSerializer를 만들어 items를, ItemDetailSerializer를 만들어 reviews field를 추가하고, view에서 action이 retrieve일 경우 DetailSerializer를 사용하도록 설정합니다.

Category retrieve api로 예를 들어 보겠습니다.

DetailSerializer, ~Serializer등 여러 개의 클래스를 생성하면 serializers.py 파일이 너무 커지기 때문에 category_serializers.py로 분리했습니다.

이전에 생성한 ItemSerializer클래스를 불러와 사용하면 ModelSerializer내부에서 items라는 attribute를 찾아 items field에 넣어줍니다.

class CategoryDetailSerializer(serializers.ModelSerializer):
    items = ItemSerializer(many=True)
    class Meta:
        model = Category
        fields = '__all__'

CategoryViewSet에서는 retrieve인 경우 DetailSerializer를 사용하도록 수정합니다.

ModelViewSet의 코드를 까보면 내부적으로 get_serializer_class라는 함수를 호출해 serializer를 가져옵니다.

따라서 해당 함수를 overwrite해주면 됩니다.

class CategoryViewSet(viewsets.ModelViewSet):
    queryset = Category.objects.all()
    serializer_class = CategorySerializer

    def get_serializer_class(self):
        if self.action == 'retrieve':
            return CategoryDetailSerializer
        return CategorySerializer

실제로확인해보면 아래와 같이 잘 나오는것을 볼 수 있습니다.

Permission 적용

category, item항목은 유저가 직접 편집하기에 위험부담이 있습니다.

따라서 해당 항목들은 admin유저만이 편집할 수 있도록 권한을 제어하려 합니다.

DRF에서 Permission을 손쉽게 제어하는 방식을 소개하고 있으니, 이를 활용하면 됩니다.

Permissions

우리는 DjangoModelPermissionsOrAnonReadOnly 라는 permission_class 를 사용합니다.

views에서 IsAdminUser와 DjangoModelPermissionsOrAnonReadOnly를 활용해 permission_classes를 작성합니다.

class CategoryViewSet(viewsets.ModelViewSet):
    queryset = Category.objects.all()
    serializer_class = CategorySerializer
    permission_classes = [DjangoModelPermissionsOrAnonReadOnly]

이제 testcode를 다시 실행해보면 테스트가 실패하는것을 볼 수 있습니다.

category, item의 수정 권한이 admin으로 변경되었기 때문에 해당 api를 이용하기 위해선 admin user로 로그인 해야합니다.

category api testcode를 예로 들겠습니다.

아래와 같이 setUp함수에서 권한을 가진 user를 생성하고, 테스트시에 해당 user로 로그인하도록 코드를 수정합니다.

def setUp(self) -> None:
    self.user = self.make_user(perms=['reviews.add_category', 'reviews.change_category', 'reviews.delete_category'])
    self.category = Category.objects.create(name='test category')
    self.category2 = Category.objects.create(name='test category 2')

...

def test_create_category(self):
    with self.login(username=self.user.username):
        data = {'name': 'test category 3'}
        response = self.post('/reviews/categories/', data=data)
        self.assert_http_201_created(response)
        self.assertEqual(response.data['name'], data['name'])
        self.assertEqual(Category.objects.count(), 3)
        self.assertEqual(Category.objects.last().name, data['name'])

다음으로 각 item에 대한 review는 로그인을 한 사람만이 작성할 수 있고, 해당 리뷰를 작성한 사람만이 수정할 수 있도록 제한합니다.

이번 단계에선 Object level permissions 을 활용합니다.

permissions.py 파일을 작성합니다.

class ReviewAuthorOrReadOnly(permissions.BasePermission):
    """
    인증 시 사용자는 자신의 리뷰만 수정할 수 있습니다.
    인증되지 않은 사용자는 읽기 전용입니다.
    """
    def has_permission(self, request, view):
        if request.method in permissions.SAFE_METHODS:
            return True
        return request.user.is_authenticated

    def has_object_permission(self, request, view, obj):
        if request.method in permissions.SAFE_METHODS:
            return True
        return obj.user == request.user

ReviewViewSet에 추가합니다.

class ReviewViewSet(viewsets.ModelViewSet):
    queryset = Review.objects.all()
    serializer_class = ReviewSerializer
    permission_classes = [ReviewAuthorOrReadOnly]

다시 testcode를 실행해보면 review 부분에서 권한이슈로 테스트가 실패하는걸 확인할 수 있습니다.

review testcode에서도 로그인 코드를 추가해줍니다.

지금까지 연결된 model을 한번에 출력하고, permission을 적용해보았습니다.

다음시간엔 쿼리성능 개선 방법에 대해 알아보도록 하겠습니다.

하루만에 완성하는 Django+DRF 서비스(3) was originally published in None on Medium, where people are continuing the conversation by highlighting and responding to this story.

안녕하세요. 휴먼스케이프 june입니다.

이번 시간엔 testcode를 작성, api 생성에 대해 알아보도록 하겠습니다.

프로젝트를 진행한 코드는 깃허브에 올라가 있으니 참고하시기 바랍니다.

GitHub - Root-kjh/drf_practice

테스트코드 작성

테스트코드는 간단히 e2e테스트코드만 작성하도록 하겠습니다.

개인적으로는 e2e테스트를 작성하고, 추가적으로 배포환경에서 이슈가 발생했을때 해당 케이스를 테스트코드로 작성 후 해결해나가며 테스트코드를 늘리는 쪽으로 작업합니다.

(물론 dev, staging phase에서 테스트를 진행하며 나오는 이슈도 테스트코드로 작성합니다.)

테스트코드를 작성하기 전, python답게 테스트코드를 효율적으로 작성할 수 있도록 도와주는 훌륭한 패키지를 추가해줍니다.

poetry add django-test-plus

이제 카테고리 crud(create/read/update/delete)에 대한 테스트코드를 추가해줍시다.

from test_plus.test import APITestCase

from .models import Category

class CategoryApiTestCase(APITestCase):
    def setUp(self) -> None:
        self.category = Category.objects.create(name='test category')
        self.category2 = Category.objects.create(name='test category 2')

    def test_get_categories(self):
        response = self.get('/reviews/categories/')
        self.assert_http_200_ok(response)
        self.assertEqual(len(response.data), 2)
        self.assertEqual(response.data[0]['name'], self.category.name)
        self.assertEqual(response.data[1]['name'], self.category2.name)

    def test_get_category(self):
        response = self.get(f'/reviews/categories/{self.category.id}/')
        self.assert_http_200_ok(response)
        self.assertEqual(response.data['name'], self.category.name)

    def test_create_category(self):
        data = {'name': 'test category 3'}
        response = self.post('/reviews/categories/', data=data)
        self.assert_http_201_created(response)
        self.assertEqual(response.data['name'], data['name'])
        self.assertEqual(Category.objects.count(), 3)
        self.assertEqual(Category.objects.last().name, data['name'])

    def test_update_category(self):
        data = {'name': 'test category 3'}
        response = self.put(f'/reviews/categories/{self.category.id}/', data=data)
        self.assert_http_200_ok(response)
        self.assertEqual(response.data['name'], data['name'])
        self.assertTrue(Category.objects.filter(name=data['name']).exists())
        self.assertFalse(Category.objects.filter(name=self.category.name).exists())

    def test_delete_category(self):
        response = self.delete(f'/reviews/categories/{self.category.id}/')
        self.assert_http_204_no_content(response)
        self.assertFalse(Category.objects.filter(name=self.category.name).exists())
        self.assertEqual(Category.objects.count(), 1)

아주 간단한 테스트코드를 작성했습니다.

테스트를 실행 python3 manage.py test 하게되면 당연히 실패합니다.

이제 api를 작성해 해당 테스트코드가 동작하도록 만들어주면 됩니다.

API 작성

drf의 구성 요소는 크게 model, serializer, view, url이 있으며, 필요하다면 filters, permissions등의 추가 요소가 사용됩니다.

model은 이미 작성했으니 serializer, view, url을 작성해 api를 작성해봅시다.

serializer는 django orm에서 쿼리의 리턴으로 사용하는 queryset , model instance등을 json등의 형식으로 변경해주고, 반대로 json형식을 model instance 등으로 변환해줍니다.

ModelSerializer를 상속한 serializer 클래스를 만들고, 사용할 모델을 model 에 넣어주면 해당 모델의 필드를 자동으로 인식해 serializer를 만들어줍니다.

from rest_framework import serializers
from .models import Review, Category

class CategorySerializer(serializers.ModelSerializer):
    class Meta:
        model = Category
        fields = "__all__"

이제 해당 serializer를 사용해 view를 만들어봅시다.

drf 문서를 보면 여러가지 view 종류가 있는데, 마트에서 쇼핑하듯 입맛에 맞는 클래스를 사용하면 됩니다.

저는 특별히 커스텀할 요소가 없으면 model에 맞게 기본적인 crud api를 만들어주는 ModelViewSet을 사용하도록 하겠습니다.

ModelView을 상속받고, queryset, serializer_class를 지정해주면 crud api가 생성됩니다.

from rest_framework import viewsets
from .models import Category
from .serializers import CategorySerializer

class CategoryViewSet(viewsets.ModelViewSet):
    queryset = Category.objects.all()
    serializer_class = CategorySerializer

이제 만든 view를 특정 path에서 사용할 수 있도록 path와 view를 연결해줍니다.

django에서는 보통 urls.py에서 해당 로직을 처리하게 됩니다.

방금 만든 viewset을 처리하는 router를 만들고, 이를 urlpatterns에 넣으면 django에서 인식하게 됩니다.

from .views import CategoryViewSet
from rest_framework import routers
from django.urls import path, include

router = routers.DefaultRouter()
router.register(r'categories', CategoryViewSet)

urlpatterns = [
    path('', include(router.urls)),
]

만든 urls.py를 기본 앱(settings.py가 위치한 바로 그 앱)의 urls에서도 연결시켜줍니다.

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('reviews/', include('reviews.urls')),
]

이제 django -> main urls.py -> reviews urls.py로 연결되어 categories api가 생성됩니다.

테스트코드를 다시 실행시켜보면 모두 성공하는것을 확인할 수 있습니다.

또, python manage.py runserver를 실행한 뒤 http://127.0.0.1:8000/reviews/categories/에 접속하면 실제로 api를 테스트해볼 수 있습니다.

Django admin

제게 Django의 최대 장점을 두가지 꼽으라면 높은 생산성과 자동 생성되는 어드민 페이지를 대답할 것 같습니다.

그만큼 Django admin은 매우 간편하게 사용할 수 있고, 사용하기에도 매우 용이합니다.

얼마나 간편하냐면, 단 1줄의 코드로 category의 어드민을 만들 수 있을 정도입니다.

이제 그 방법에 대해 알아보도록 하겠습니다.

python manage.py createsuperuser 명령어를 입력해 admin페이지에서 사용할 superuser를 생성합니다.

http://127.0.0.1:8000/admin/에 접속해 아까 생성한 계정정보를 입력하면

admin page를 확인할 수 있습니다.

지금은 아무 코드를 작성하지 않아 user, group만 관리할 수 있는 admin page입니다.

다시 코드로 돌아와, reviews/admin.py에 category, item, review model을 등록해줍시다.

from django.contrib import admin

from .models import Category, Item, Review

admin.site.register(Category)
admin.site.register(Item)
admin.site.register(Review)

(총 7줄이지만, category 부분만 보면 1줄… 1줄입니다.)

다시 어드민 페이지에 접속하면 다음과 같이 각 모델에 대한 탭이 생성된걸 확인할 수 있습니다.

category의 crud를 작성했으니, 이를 조금 응용해서 Item, review의 crud를 작성해주시면됩니다.(귀찮아서 스킵..)

다음 시간엔 category, Item, review의 모델 연결관계를 api에서 표현하는 방법, permission을 적용하는 방법(일반유저가 items를 삭제/수정하지 못하도록 관리)에 대해 알아보도록 하겠습니다.

하루만에 완성하는 Django+DRF 서비스(2) was originally published in None on Medium, where people are continuing the conversation by highlighting and responding to this story.

안녕하세요. 휴먼스케이프 june입니다.

이번 시리즈에서는 django+drf 조합을 이용해 서비스를 빠르게 만드는 법에 대해 알아보도록 하겠습니다.

이번 시간엔 간단히 프로젝트의 생성, 모델의 생성만 진행할 예정입니다.

프로젝트를 진행한 코드는 깃허브에 올라가 있으니 참고하시기 바랍니다.

GitHub - Root-kjh/drf_practice

주제 선정

어쩌면 이번 시리즈에서 가장 어려울 수도 있는(?) 부분입니다.

저는 이전부터 만들고 싶었던 모든것을 리뷰해 정보의 비대칭성을 개선 하는 서비스를 주제로 잡았습니다.

서비스의 이름은 직관적으로 review_all 이라고 하겠습니다.

글을 읽으며 같은 서비스를 만들어도 좋고, 다른 서비스를 만들어가도 좋습니다.

파이썬 설치

자연어 처리를 하기 위해선 파이썬을 설치해야 합니다.

아래 사이트에 접속해 파이썬을 설치해줍니다.

Welcome to Python.org

Django 프로젝트 생성

django project를 생성하기 위해 python에 django를 설치해줍니다.

$ python -m pip install Django

django project를 생성해줍니다.

$ django-admin startproject django_practice

패키지 버전 관리(Poetry)

파이썬 패키지 버전 관리는 poetry를 사용해서 진행합니다.

아래 문서를 보고 poetry install을 진행합니다.

Introduction

poetry install이 완료되었으면 poetry init 명령어를 실행해 pyproject.toml 파일을 생해줍니다.

새로 생성된 pyproject.toml에는 django package가 포함되어있지 않으니 django를 추가합니다.

$ poetry add django

django+drf 조합을 사용할 예정이기 때문에 drf도 미리 설치해줍니다.

poetry add djangorestframework

Home - Django REST framework

settings.py -> INSTALLED_APPS에 rest_framework를 추가해 줍니다.

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'rest_framework',
]

poetry의 python virtualenv에 들어가려면 다음 명령어를 입력하면 됩니다. poetry shell

Reviews 앱 생성

review_all은 크게 3가지 모델로 구성할 예정입니다.

모델: 일반적으로 database의 table과 연결되며, 데이터의 구조와 동작을 표현합니다.

1. 리뷰할 item들을 그룹화한 category
2. 리뷰할 item들을 저장하는 item
3. 리뷰를 저장하는 review

먼저 reviews앱을 생성해줍니다.

python3 manage.py startapp reviews

settings.py -> INSTALLED_APPS에 방금 생성한 review를 추가해 줍니다.

이제 reviews/models.py에 각 모델을 생성합니다.

from django.db import models
from django.contrib.auth.models import User

class Category(models.Model):
    name = models.CharField(max_length=100)

    def __str__(self):
        return self.name


class Item(models.Model):
    name = models.CharField(max_length=100)
    category = models.ForeignKey(Category, on_delete=models.CASCADE, related_name='items')
    description = models.TextField(null=True, blank=True)
    image = models.ImageField(upload_to='images/', null=True)
    date = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.name


class Review(models.Model):
    item = models.ForeignKey(Item, on_delete=models.CASCADE, related_name='reviews')
    context = models.TextField()
    created = models.DateTimeField(auto_now_add=True)
    updated = models.DateTimeField(auto_now=True)
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='reviews')

    def __str__(self):
        return self.name

이렇게 django+drf 프로젝트의 생성, 모델 작성까지 빠르게 진행했습니다.

다음 시간엔 testcode를 작성하고 api를 생성해보도록 하겠습니다.

하루만에 완성하는 Django+DRF 서비스(1) was originally published in None on Medium, where people are continuing the conversation by highlighting and responding to this story.

안녕하세요. 휴먼스케이프 june입니다.

이번시간엔 임상연구 업데이트 시 변경 부분을 감지하고, 해당 부분만을 업데이트하는 방법에 대해 알아보겠습니다.

프로젝트를 진행한 코드는 깃허브에 올라가 있으니 참고하시기 바랍니다.

GitHub - Root-kjh/django_practice

실제 서비스하는 임상연구는 유료 번역api를 사용해 번역의 퀄리티를 높이기 때문에 관련 비용이 굉장히 많이 나오게 되고, 번역 태스크에서의 병목현상이 많이 일어나기 때문에 이 부분을 최적화하는게 굉장히 중요합니다.

진행 과정은 다음과 같습니다.

1. convert태스크에서 변경된 부분을 확인합니다.

study model의 필드는 <updated>와 같은 값으로 채워 변경됨을 알리고, 연결된 모델은 인스턴스를 특정할 수 없는 고유 키가 없기 때문에(clinicaltrials에서 id값을 제공하지 않음) 모든 값이 같을 경우에만 유지, 하나라도 다를 경우 생성/제거합니다.

2. translate태스크에서 변경된 부분(marked study model field, created related model instance)에 대한 번역을 새로 생성합니다.

이 때 제거된 related model instance의 번역본의 경우, 이미 on_delete 규칙에 따라 convert 태스크에서 같이 제거되었기 때문에 신경쓰지 않아도 됩니다.

먼저, 간단하게 구현할 수 있는 study model의 필드 변경 감지 코드를 작성해보겠습니다.

번역필드만 확인하면 되기 때문에 TRANSLATE_FIELDS로 지정해두고 사용하도록 하겠습니다.

def mark_updated_sutdy_field(study, convert_data):
    TRANSLATE_FIELDS = ['title', 'overall_status', 'phase']

    translated_studies = Study.objects.filter(translate_from_study=study)
    for field in TRANSLATE_FIELDS:
        if getattr(study, field) != convert_data[field]:
            for translated_study in translated_studies:
                setattr(translated_study, field, "<updated>")
    for translated_study in translated_studies:
        translated_study.save()

이제 related model instance에서 기존 instance를 재활용하도록 수정해줍니다.

def convert_conditions(condition_module):
    conditions = []
    condition_instances = list(Condition.objects.filter(name__in=condition_module))
    for condition in condition_module:
        condition_id = None
        for condition_instance in condition_instances:
            if condition_instance.name == condition:
                condition_id = condition_instance.id
                break
        conditions.append({
            'id':  condition_id,
            'name': condition,
            'locale': 'en',
        })
    return conditions

def convert_interventions(study, intervention_module):
    interventions = []
    intervention_instances = list(Intervention.objects.filter(study=study))
    for intervention in intervention_module:
        intervention_id = None
        for intervention_instance in intervention_instances:
            if (
                intervention_instance.name == intervention.get('InterventionName', None) and \
                intervention_instance.intervention_type == intervention.get('InterventionType', None) and \
                intervention_instance.description == intervention.get('InterventionDescription', None)
            ):
                intervention_id = intervention_instance.pk
                intervention_instances.remove(intervention_instance)
                break
        interventions.append({
            'id': intervention_id,
            'intervention_type': intervention.get('InterventionType', None),
            'name': intervention.get('InterventionName', None),
            'description': intervention.get('InterventionDescription', None),
            'locale': 'en',
        })
    return interventions

def convert_eligibilities(study, eligibility_module):
    if eligibility_module is None:
        return []
    eligibility_instances = list(Eligibility.objects.filter(study=study))
    eligibility_id = None
    for eligibility_instance in eligibility_instances:
        if (
            eligibility_instance.gender == eligibility_module.get('Gender', None) and \
            eligibility_instance.minimum_age == eligibility_module.get('MinimumAge', None) and \
            eligibility_instance.maximum_age == eligibility_module.get('MaximumAge', None) and \
            eligibility_instance.healthy_volunteers == eligibility_module.get('HealthyVolunteers', None) and \
            eligibility_instance.criteria == eligibility_module.get('EligibilityCriteria', None)
        ):
            eligibility_id = eligibility_instance.pk
            eligibility_instances.remove(eligibility_instance)
            break
    return [{
        'id': eligibility_id,
        'gender': eligibility_module.get('Gender', None),
        'minimum_age': eligibility_module.get('MinimumAge', None),
        'maximum_age': eligibility_module.get('MaximumAge', None),
        'healthy_volunteers': eligibility_module.get('HealthyVolunteers', None),
        'criteria': eligibility_module.get('EligibilityCriteria', None),
        'locale': 'en',
    }]

convert 태스크를 실행하면 변경사항이 없는 부분은 재활용하고, 변경된 부분만 인식함을 확인할 수 있습니다.

이제 translate 태스크를 수정합니다.

1. trnaslated_studies를 제거하는 코드 삭제
2. related model instance중 translated instance가 없는 부분만 번역하도록 수정
3. <updated>마크 되어있는 study model컬럼만 다시 번역하도록 수정

1번은 다들 금방 할 수 있기 때문에 2, 3번에 대해서만 설명하도록 하겠습니다.

• translated_study 인스턴스를 업데이트 하는 방식으로 진행하기 때문에, translate 태스크 진행 시 translated_study 인스턴스를 찾아 translate_study함수에 인자로 넣어줍니다.
• translate_study함수는 인자로 받은 translated_study(default: None)에서 <update>마크가 되어있는 부분만 새로 번역하고, 나머지는 기존 데이터를 그대로 사용합니다.
• intervention, condition 등의 연결된 모델들은 translated object가 있는지 확인하고, 없으면 생성해 사용합니다.

코드를 보면 다음과 같습니다.

def translate_study(study, translated_study=None):
    """
    임상 연구 데이터를 번역하는 메소드
    """
    translated_text_dict = {translate_field:None for translate_field in TRANSLATE_FIELDS}
    if translated_study is not None:
        for translate_field in TRANSLATE_FIELDS:
            if getattr(translated_study, translate_field) != "<updated>":
                translated_text_dict[translate_field] = getattr(translated_study, translate_field)
    for translate_field in TRANSLATE_FIELDS:
        if translated_text_dict[translate_field] is None and getattr(study, translate_field) is not None:
            translated_text_dict[translate_field] = translate(getattr(study, translate_field))

    conditions = []
    for condition in study.conditions.all().prefetch_related(Prefetch("translated_conditions", queryset=Condition.objects.filter(locale='ko'))):
        translated_condition = condition.translated_conditions.first()
        if translated_condition is not None:
            conditions.append({
                'id': translated_condition.pk,
                'name': translated_condition.name,
                'locale': 'ko',
                'translate_from_condition': condition.pk
            })
        else:
            conditions.append({
                'name': translate(condition.name),
                'locale': 'ko',
                'translate_from_condition': condition.pk
            })

    interventions = []
    for intervention in Intervention.objects.filter(study=study).prefetch_related(Prefetch("translated_interventions", queryset=Intervention.objects.filter(locale='ko'))):
        translated_intervention = intervention.translated_interventions.first()
        if translated_intervention is not None:
            interventions.append({
                'id': translated_intervention.pk,
                'intervention_type': translated_intervention.intervention_type,
                'name': translated_intervention.name,
                'description': translated_intervention.description,
                'locale': 'ko',
                'translate_from_intervention': intervention.pk
            })
        else:
            interventions.append({
                'intervention_type': translate(intervention.intervention_type),
                'name': translate(intervention.name),
                'description': translate(intervention.description),
                'locale': 'ko',
                'translate_from_intervention': intervention.pk
            })

    eligibilities = []
    for eligibility in Eligibility.objects.filter(study=study).prefetch_related(Prefetch("translated_eligibilities", queryset=Eligibility.objects.filter(locale='ko'))):
        translated_eligibility = eligibility.translated_eligibilities.first()
        if translated_eligibility is not None:
            eligibilities.append({
                'id': translated_eligibility.pk,
                'gender': translated_eligibility.gender,
                'minimum_age': translated_eligibility.minimum_age,
                'maximum_age': translated_eligibility.maximum_age,
                'healthy_volunteers': translated_eligibility.healthy_volunteers,
                'criteria': translated_eligibility.criteria,
                'locale': 'ko',
                'translate_from_eligibility': eligibility.pk
            })
        else:
            eligibilities.append({
                'gender': translate(eligibility.gender),
                'minimum_age': eligibility.minimum_age,
                'maximum_age': eligibility.maximum_age,
                'healthy_volunteers': translate(eligibility.healthy_volunteers),
                'criteria': translate(eligibility.criteria),
                'locale': 'ko',
                'translate_from_eligibility': eligibility.pk
            })

    return {
        'nct_id': study.nct_id,
        'title': translated_text_dict['title'],
        'results_first_submitted_date': study.results_first_submitted_date,
        'last_update_submitted_date': study.last_update_submitted_date,
        'start_date': study.start_date,
        'completion_date': study.completion_date,
        'overall_status': translated_text_dict['overall_status'],
        'phase': translated_text_dict['phase'],
        'enrollment': study.enrollment,
        'interventions': interventions,
        'conditions': conditions,
        'eligibilities': eligibilities,
        'locale': 'ko',
        'translate_from_study': study.pk,
        'control_status_type': ControlStatusType.COMPLETED,
    }

이렇게 임상연구 적재, 번역, 간단한 최적화까지 빠르게 진행해보았습니다.

여기서 조금만 더 응용한다면 다국어 번역, 여러 데이터 소스에서 임상연구 데이터 적재, 더 좋은 최적화 등 개선할 부분이 많으니 시도해보셔도 좋을듯 합니다.

Django 데이터 적재 시리즈는 여기서 마무리 짓고, 이후 하루만에 완성하는 Django+DRF 서비스시리즈로 찾아오도록 하겠습니다.

처음부터 시작하는 Django 데이터 적재(5) — END was originally published in None on Medium, where people are continuing the conversation by highlighting and responding to this story.