CDK란 무엇일까요?

CDK는 AWS Cloud Development Kit의 약어로, AWS 인프라 관리를 개발자가 사용하는 언어(예: Python, Java, TypeScript 등)로 관리할 수 있게 해주는 도구입니다.

핵심만 말하자면

내가 사용하는 언어로 AWS 인프라 관리를 도와주는 도구

누구에게 필요할까요

개발자가 AWS 인프라 관리를 자신이 사용하는 언어로 관리하고 싶을 때, CDK를 사용할 수 있습니다. 특히 TypeScript를 사용하는 경우 프론트엔드, 백엔드, 인프라 관리 등 다양한 작업을 동시에 할 수 있습니다.

CDK는 어떤 문제를 해결해줄 수 있을까요

• 우선 laC의 장점과 동일한 부분은 간단하게만 적겠습니다
  • 일관성:
    • IaC를 사용하면 코드로 인프라를 구성하기 때문에 모든 인프라 구성이 코드화 가능.
    • 반복적인 인프라 작업을 자동화.
  • 효율성:
    • 코드 재사용성과 모듈화가 가능. 또한,
    • 코드 버전 관리, 변경 내역 추적, 빠른 배포와 롤백 가능
    • 작업 시간 단축 = 개발자 행복
  • 안정성:
    • 자동화된 테스트와 검증 절차를 통해 오류 감지, 오류 방지 가능.

이거를 이제 내가 잘 쓰는 언어로 할 수 있다는 건데 유명한 terraform 같은 경우에도 새로 언어를 배워한다는게, 특히 스타트업 입장에서는 러닝커브라는건 시간적인 리스크가 있어서 "이렇게 할바에는 그냥 수동으로 하는게 빠르겠다. 도입 포기!" 이런 결과로 가기 쉽다는 거죠.

요약을 하자면 IaC 도입을 가장 쉽게 할 수 있는 방법인거 같아요

예시

import * as cdk from 'aws-cdk-lib';
import * as ec2 from 'aws-cdk-lib/aws-ec2';
import * as ecs from 'aws-cdk-lib/aws-ecs';
import * as ecsPatterns from 'aws-cdk-lib/aws-ecs-patterns';

const app = new cdk.App();
const stack = new cdk.Stack(app, 'MyStack');

const vpc = new ec2.Vpc(stack, 'MyVpc', { maxAzs: 2 });

const cluster = new ecs.Cluster(stack, 'MyCluster', { vpc });
const fargateService = new ecsPatterns.ApplicationLoadBalancedFargateService(stack, 'MyFargateService', {
  cluster,
  memoryLimitMiB: 512,
  cpu: 256,
  taskImageOptions: {
    image: ecs.ContainerImage.fromRegistry('amazon/amazon-ecs-sample'),
  },
});

new cdk.CfnOutput(stack, 'LoadBalancerDNS', { value: fargateService.loadBalancer.loadBalancerDnsName });

위 코드는 AWS CDK를 사용해 Amazon ECS 클러스터와 Fargate 서비스를 생성하는 TypeScript 코드 예시입니다.

어떻게 쓰고 계시나요

지금 적용되어있는 형태

• lambda + apigateway Restful API
• codepipeline
• vpc
• elastic cache
• eventbridge(스케쥴러용)

단점은 없나요

CDK를 사용하면 인프라 관리가 코드 형태로 되므로, IaC(Infrastructure as Code)를 사용하지 않을 때 발생할 수 있는 문제들을 방지할 수 있습니다. 그러나 CDK 역시 일정한 학습 곡선이 필요하고, AWS 서비스가 업데이트되면서 CDK에서 지원되지 않는 경우도 있습니다.

실제로 제 경험상, API Gateway에서 리소스를 삭제하고 코드를 변경 후 배포하면, 삭제된 리소스가 남아있어 배포가 실패할 수 있습니다. 또한, 많은 리소스를 한 번에 변경하려다가 실패하여 롤백하는 경우에는 롤백 시간이 길어지는 등의 문제가 발생할 수 있습니다.

하지만 이러한 문제들은 CDK의 원리와 사용 방법을 숙지하고 있으면 충분히 해결할 수 있습니다.

결론

CDK는 개발자가 AWS 인프라를 코드로 관리할 수 있게 해주는 도구로, 코드 재사용성과 모듈화가 가능하며, 코드 버전 관리와 변경 내역 추적, 빠른 배포와 롤백이 가능하다는 장점이 있습니다. 또한, CDK를 사용하면 AWS 인프라 관리에 대한 이해도가 높아질 수 있다는 점도 큰 장점입니다.

그러나 AWS 서비스가 업데이트되면서 CDK에서 지원되지 않는 경우도 있으므로, CDK를 사용하기 전에 AWS 공식 문서를 꼼꼼하게 확인하는 것이 중요합니다.

시간이 날때마다 그동안 적용해왔던 경험을 시리즈로 공유해보겠습니다. 감사합니다.

k8s 환경 안에서 app이 제대로 동작되지 않는 것으로 보입니다. 원인을 찾아 복구해 보겠습니다.

kubectl get all --all-namespaces

describe로 좀 더 자세히 보겠습니다.

kubectl describe pod -n kube-system kube-scheduler-controlplane

이부분에서 문제가 있는 것 같습니다.

kube-scheduler의 manifestfile을 수정하도록하겠습니다.

vi /etc/kubernetes/manifests/kube-scheduler.yaml 

# in kube-scheduler.yaml
spec:
  containers:
  - command:
    - kube-scheduler
    - --authentication-kubeconfig=/etc/kubernetes/scheduler.conf
    - --authorization-kubeconfig=/etc/kubernetes/scheduler.conf
    - --bind-address=127.0.0.1
    - --kubeconfig=/etc/kubernetes/scheduler.conf
    - --leader-elect=true
    - --port=0

다시 체크해서 복구된 것을 확인합니다.

kubectl get all --all-namespaces

good~

2

이번에는 deployement의 pod를 1개에서 2개로 scale out 한 이후 이상현상이 발생했습니다.

kubectl get all --all-namespaces

자세히 보겠습니다.

kubectl describe pod -n kube-system kube-controller-manager-controlplane 

그러나 descibe 만으로는 원인을 파악하기 어려웠습니다. 로그를 보겠습니다.

kubectl logs -n kube-system kube-controller-manager-controlplane  

config file 이름에 정확하지 않은 것으로 판단됩니다.

root@controlplane:~# ls /etc/kubernetes/          
admin.conf               kubelet.conf             pki/                     
controller-manager.conf  manifests/               scheduler.conf   

controller-managet.conf로 정정해줍니다.

 vi /etc/kubernetes/manifests/kube-controller-manager.yaml 

spec:
  containers:
  - command:
    ...중략...
    - --kubeconfig=/etc/kubernetes/controller-manager.conf
    ...중략...

다시 잘돌아가는지 확인해줍니다.

kubectl get all --all-namespaces

good~

3

kubectl logs -n kube-system kube-controller-manager-controlplane  

로그를 보면 무엇인가 또 경로에 문제가 있어보입니다.

kubectl describe pod -n kube-system kube-controller-manager-controlplane 

이상한 경로가 포함되어있는 것을 확인하였습니다. 올바른 경로로 수정합니다.

 vi /etc/kubernetes/manifests/kube-controller-manager.yaml 

  volumes:
  ...중략...
  - hostPath:
      path: /etc/kubernetes/pki
      type: DirectoryOrCreate
    name: k8s-certs
  ...중략...

확인해보자

good!

TL;DR

• log를 킹갓 AWS에 맡겨볼겁니다.
• watchtower를 사용할 겁니다.
• flask에 적용할겁니다.(다른 framework면 미리 죄송합니다.)

들어가며

회사에서는 로그를 특정 파일에 남긴 뒤 해당 파일을 s3의 올리는 방식을 사용하고 있습니다. 원하는 로그를 검색도 하기도 어렵고 가독성도 매우 떨어져서 "이게 맞나..." 싶었습니다. 또 로그를 확인하기 위해서 s3에서 파일을 다운받고 에디터로 열고 검색하고 하는 과정이 심히 불편쓰.. 그러다보니 로그가 있어도 잘 이용을 하지 않는 느낌을 많이 받아서 고민하던 중 간단하게 적용할 수 있는 라이브러리가 있어서 회사 적용해보았더니 기존에 비해 편리하다는 느낌을 많이 받아서 정리 겸 공유 해보려고 합니다.

IAM 사용자 생성

사용자 생성

'AWS Console - IAM - 사용자'에서 '사용자 추가'를 해주세요. *'AWS 액세스 유형 선택'에서 '엑세스 키-프로그래밍 방식 액세스' 선택 해주세요.

권한 부여

watchtower에서 다음과 같이 적절한 권한을 추가하도록 권장하고 있습니다.

we recommend that you use the arn:aws:iam::aws:policy/AWSOpsWorksCloudWatchLogs managed policy, which has just the right permissions without being overly broad. -> 지나치게 광범위하지 않으면서 적절한 "AWSOpsWorksCloudWatchLogs" 정책을 사용하는 것을 추천합니다.

IAM 생성이 완료된 모습입니다. 액세스 키와 비밀 액세스 키를 잘 가지고 있어 주세요.

watchtower 추가

pip install --upgrade pip
pip install watchtower

끝입니다.

handler 적용

# app.py
import logging

from flask import Flask, current_app
import watchtower

app = Flask(__name__)

handler = watchtower.CloudWatchLogHandler(
        log_group_name="my-service",
        log_stream_name="dev", 
)

formatter = logging.Formatter("[%(levelname)s] %(message)s")
handler.setFormatter(formatter)
handler.setLevel(logging.INFO)

app.logger.addHandler(handler)

@app.route("/")
def hello_world():
    current_app.logger.info("It`s working!!")
    return "<p>Hello, World!</p>"

적용을 위한 간단한 flask app을 만들어보았습니다.

그럼 테스트를 한번 해볼까요?

export FLASK_APP=app.py
flask run

이런... 에러가 나오네요.

botocore.exceptions.NoRegionError: You must specify a region.

이때! 1번에서 생성한 IAM 액세스 키를 추가해주어야합니다.

export AWS_DEFAULT_REGION=default_region
export AWS_ACCESS_KEY_ID=your_access_key_id
export AWS_SECRET_ACCESS_KEY=your_secret_access_key

AWS_DEFAULT_REGION = default region을 설정해주어야합니다. 서울은 'ap-northeast-2'입니다. AWS_ACCESS_KEY_ID = IAM 생성 시 받은 액세스 키를 넣어줍니다. AWS_SECRET_ACCESS_KEY = IAM 생성 시 받은 비밀 액세스 키를 넣어줍니다.

적용 후 다시 flask run 하면 이번에는 정상동작하는 것을 볼 수 있습니다.

확인하기

API를 호출 해볼까요? http://localhost:5000/

호출 후 AWS Console로 가보면 다음과 같이 코드에 설정한 로그 그룹과 로그 스트림 정보가 나타나게됩니다.

로그 전송 시간이 있어서 1분정도 딜레이가 있을 수 있습니다. 물론 해당 시간은 로그핸들러 생성시 조정할 수 있는 인자가 있습니다.

짠! log가 정상적으로 들어왔네요~ 이제 cloudwatch에서 서비스의 로그를 확인할 수 있습니다.

실제 운영 시 고려한 점

• 로그스트림에는 개발환경(dev, staging, prod)를 넣어서 구분했다.
• 실제로는 환경변수가 아닌 aws credential을 사용했다.(보안)
• 기존에 boto3를 사용하고 있었으면 boto3.client("logs")로 credential을 부여할 수 있는 인자기 있다.(라이브러리가 boto3 기반)
• 핸들러 생성시 aws credential이 적절하지 않으면 서버가 구동이 안되기 때문에 적용하지 않도록 에러 처리를 했다. (예: CI 테스트 서버)

마치며

서비스를 운영하면서 당장의 대응만 생각하다보면 로그 구축은 생각보다 우선순위에서 멀어지게 되는거 같습니다. 때문에 미래를 대비하기 위해서 읽기 쉽고, 사용하기 쉬운 로깅에 대해서 고민이 필요할 것 같습니다.

Reference

watchtower github

그렇기 때문에 회사에서 convention에 신경을 쓰는 편입니다. 하나 딱! 정해놓고 평생동안 울궈먹거나 이것을 기준삼아 발전시키기 위해 개인적으로 정리해보았습니다.

Commit Formats

Default

<type>(<optional scope>):<subject>empty separator line<optional body>empty separator line<optional footer>

• subject와 body에는 한글/영어를 혼용이 가능한 것으로 하겠습니다.
• 혹시 팀원 전원이 찬성한다면 subject와 body 역시 영어를 사용하는 것을 권장합니다.

Types

Type을 통해 해당 커밋의 가장 기본적을 목적을 파악합니다.

• feat : 기획과 관련한 기능 추가 및 삭제
• fix  : 버그 수정 및 기존 기능 수정
• refactor : 리팩토링 (기능 수정 X) * 기능 수정과 리팩토링 구분해서 올려주세요.
  • perf : 성능 향상을 위한 리팩토링
• style : code style 변경 (white-space, formatting, missing semi-colons, etc)
• test : 테스트 추가
• docs : 문서 추가/변경 (ReadMe.md)
• build: 빌드 관련 변경 (build tool, ci pipeline, dependencies, project version)
• chore : 그외 변경 ( .gitignore )

Scope

scope는 하나의 repo에 다양한 앱이 있을 경우 구분을 짓기 위함입니다.

• request과 response이 있다면 fix(request): fix typo, feat(response): fix typo
• 비지니스 로직을 구분하기 위해서 사용

Subject

subject에서는 좀더 명확한 변경 내용을 담습니다.

• required
• 59자 이하로 작성해주세요.
• 추가, 변경, 수정, 삭제, 개선 등 명사형 문장으로 작성
• 영문의 경우 명령형으로 작성
• 끝에 온점(.) 생략
• 파일명이나 함수명은 사용 X (코드에서 보여줄것임)
• ex) oo 기능 추가, 불필요한 코드 삭제

Body

• optional
• 한 줄에 59자 이하
• 추가, 변경, 수정, 삭제, 개선 등 명사형 문장으로 작성
• subject 외 자세한 내용이 필요하다면 작성

Examples

• feat: 엄청난 버튼 추가
• fix: <어떤 이유>인한 oo 컴포넌트 버그 수정
• fix: 오타 수정
• build: package.json dependency 업데이트
• refactor: oo 컴포넌트 구조 리팩토링
• style: white space 제거

Todo

• BREAKING CHANGES나 이슈번호를 남기는 방법에 대해서 이해하기
• commitlint 적용하기
• 좀더 실무적인 예시 추가하기

References

• Conventional Commit Messages

회사에서 python code formatte로 Black을 사용하고 있어 정리해보려고 합니다

왜 코드 스타일을 자동화 해야할까요?

마치 한사람이 만든 것과 같은 코드

여러사람이 코드를 짜다보면 저마다 자신이 편한 코드 스타일이 있습니다. 어떤사람은 "double quotes" 를 선호하고, 다른 누군가는 'single quote'를 선호할 수 있습니다.

서로 다른 스타일이 적용된 코드를 보면 다른 누군가이 코드를 수정하려고 할 때 혼란이 올 수 있습니다. '어떤 스타일에 맞춰야되는 거지?'

또한, 가독성 높이기 위해서는 마치 한사람이 만든 것과 같은 코드 가 중요하다고 생각합니다. 설령, 서로 어떤 스타일로 통일 하자고 결정하고 문서를 만들어서 약속을 정하는 방법이 있지만 그것으로는 부족합니다.

문서를 매번 확인 해야하는 것은 번거롭고, 거의 그럴일은 없겠지만 만약 코드 스타일에 수정사항이 있으면 일일히 확인해서 기존의 것을 수정하기도 힘듭니다.

따라서 간단한 명령으로 코드 스타일을 통일시켜주는 방법을 찾아야할 필요가 있습니다.

Black

Black is the uncompromising Python code formatter

Black은 엄격한 파이썬 코드 포맷터 입니다.

Black gives you speed, determinism, and freedom from pycodestyle nagging about formatting. You will save time and mental energy for more important matters.

여러분은 Black을 통해 코드 스타일에 대한 잔소리로부터 자유를 얻을 수 있고, 더 중요한 문제에 시간과 에너지를 쏟을 수 있도록 해줍니다. (의역)

Black makes code review faster by producing the smallest diffs possible

Black은

Black Playground에서 설치없이 어떻게 작용하는지 확인 해볼 수 있습니다.

설치

pip install black

사용방법

• 변경전

  from seven_dwwarfs import Grumpy, Happy, Sleepy, Bashful, Sneezy, Dopey, Doc
  x = {  'a':37,'b':42,
  

'c':927}

x = 123456789.123456789E1234567891111111111111

if very_long_variable_name is not None and
very_long_variable_name.field > 0 or
very_long_variable_name.is_debug: z = 'hello '+'world' else: world = 'world' a = 'hello {}'.format(world) f = rf'hello {world}' if (this and that): y = 'hello ''world'#FIXME: https://github.com/python/black/issues/26 class Foo ( object ): def f (self ): return 37-2 def g(self, x,y=42): return y def f ( a: List[ int ]) : return 37-a[42-u : y*3] def very_important_function(template: str,*variables,file: os.PathLike,debug:bool=False,): """Applies variables to the template and writes to file.""" with open(file, "w") as f: ...

fmt: off

custom_formatting = [ 0, 1, 2, 3, 4, 5, 6, 7, 8, ]

fmt: on

regular_formatting = [ 0, 1, 2, 3, 4, 5, 6, 7, 8, ]

다음과 같이 보기만해도 불편함으로 가득한 코드가 있습니다.
해당 코드를 작성한뒤 터미널에 다음과 같이 입력해줍니다.
```bash
black {파일 또는 폴더 이름}

• 변경 후

  from seven_dwwarfs import Grumpy, Happy, Sleepy, Bashful, Sneezy, Dopey, Doc
  

x = {"a": 37, "b": 42, "c": 927}

x = 123456789.123456789e1234567891111111111111

if ( very_long_variable_name is not None and very_long_variable_name.field > 0 or very_long_variable_name.is_debug ): z = "hello " + "world" else: world = "world" a = "hello {}".format(world) f = rf"hello {world}" if this and that: y = "hello " "world" # FIXME: https://github.com/python/black/issues/26

class Foo(object): def f(self): return 37 * -2

def g(self, x, y=42):
    return y

def f(a: List[int]): return 37 - a[42 - u : y ** 3]

def very_important_function( template: str, *variables, file: os.PathLike, debug: bool = False, ): """Applies variables to the template and writes to file.""" with open(file, "w") as f: ...

fmt: off

custom_formatting = [ 0, 1, 2, 3, 4, 5, 6, 7, 8, ]

fmt: on

regular_formatting = [ 0, 1, 2, 3, 4, 5, 6, 7, 8, ]

확실히 이전 보다 깔끔해졌다고 느껴집니다.
이렇게 **Black**은 내가 어떻게 코드를 작성하든 일관된 스타일로 변경해줍니다.

다음과 같이 `fmt: on` 통해 원하는 구분에는 포맷팅이 적용되지 않게 할 수 있습니다.
```python
## 해당 구문은 오히려 포맷을 적용하지 않았을 때 가독성이 좋습니다.
# fmt: off
custom_formatting = [
    0,  1,  2,
    3,  4,  5,
    6,  7,  8,
]
# fmt: on
regular_formatting = [
    0,
    1,
    2,
    3,
    4,
    5,
    6,
    7,
    8,
]

옵션

여러 옵션을 통해 팀의 코드 스타일에 맞춰 custom 할 수 있습니다.

-l : 한 라인에 최대 글자 수 (초기값 88)

—diff : 파일을 변경하지 않고 변경되는 부분을 콘솔로 보여준다.

—color : —diff를 사용했을 때 변경점에 색을 입힌다.

black {파일명 또는 폴더명} -l 80 --diff --color

다음과 같이 사용하면 변경사항을 preview 할 수 있습니다.

참고

Black

코드

def detect_mine(ground, scope):
    size = len(ground)
    index = { 
        'x': 0, 
        'y': 0
    }
    result = 0
    while True:
        sum = 0
        # scope 내부 검사
        for x in range(index['y'], index['y'] + scope):
            for y in range(index['x'], index['x'] + scope):
                if ground[y][x]:
                    sum += 1
        # 최대값 판별
        result = max(result, sum)
        # index 변경
        if index['x'] + scope == size:
            if index['y'] + scope == size:
                return result
            index['x'] = 0
            index['y'] += 1
        else:
            index['x'] += 1


if __name__ == '__main__':

    matrix = [
                [1, 0, 0, 1, 0],
                [0, 1, 0, 0, 1],
                [0, 0, 0, 1, 0],
                [0, 0, 0, 0, 0],
                [0, 0, 1, 0, 0],
    ]
    print(detect_mine(matrix, 3))
    #3

Reference

제주 코딩 베이스 캠프 코딩 페스티벌 python 100제

개념

1. 시작위치를 q 넣고, depth를 -1로 초기화
2. depth를 1 증가
3. depth별로 loop를 진행 (for i in 0~size(q))
4. 해당 depth의 node를 pop
5. 목표점에 도달했는지 비교
6. 목표점에 도달 했다면 depth return
7. 목표점에 도달하지 못했다면 이동 경로에 node를 넣고 다음경로를 q에 push
8. 동일한 depth의 검사가 끝날 때까지 4~7 반복
9. 더이상 q가 없을 때까지 진행(while)

해당 방식은 알고리즘은 트리 구조에서만 가능하다.

코드

from collections import deque 

def shortest_path(graph, start, end):
    result = -1
    visited = []
    q = deque([start])
    while q:
        result += 1
        size = len(q)

        for _ in range(size):
            node = q.popleft()
            if node == end:
                return result

            visited.append(node)
            q += graph[node] - set(visited)

if __name__ == '__main__':
    graph = {
        'A': set(['B', 'C']),
        'B': set(['A', 'D', 'E']),
        'C': set(['A', 'F']),
        'D': set(['B']),
        'E': set(['B', 'F']),
        'F': set(['C', 'E'])}

    print(shortest_path(graph, 'A', 'F'))
    # 2

개념

1. 1~n까지 loop
2. 해당 숫자를 문자열로 분리 loop
3. 해당 글자가 0이거나 3의 배수가 아니면 return 0
4. 3번 조건이 아니면 return 1
5. 2~4을 카운트

코드

def game_369(n):

    def check_only_369(n):
        for c in str(n):
            if int(c) == 0 or int(c) % 3 != 0:
                return 0
        return 1

    cnt = 0
    for i in range(1,n+1):
        cnt += check_only_369(i) 
    return cnt 

if __name__ == '__main__':

    print(game_369(93))
    # 10

정규식으로도 풀어보자

개념

1. 정규표현식 = ^[369]+$ => 3,6,9만 포함하는 문자열
2. 해당 문자열이 정규표현식과 일치하면 카운트

코드

import re
def game_369(n):

    cnt = 0
    p = re.compile('^[369]+$')
    for i in range(1,n+1):
        if re.search(p, str(i)):
            cnt += 1 
    return cnt 

if __name__ == '__main__':

    print(game_369(93))
    # 10

정규표현식을 사용하면 조건이 있는 문자열를 판단할 때 더 간단하게 해결할 수 있다.

Reference

제주 코딩 베이스 캠프 코딩 페스티벌 python 100제

def elevator(dumps, weight):
    result = 0
    dumps.sort()

    while (dumps):
        dump = dumps.pop()
        div = weight // dump
        mod = weight % dump

        if dumps == []:
            if mod is not 0:
                return -1
            else:
                result += div
        else:
            if div:
                result += div
                weight -= div * dump                

    return result

if __name__ == '__main__':
    print(elevator([3,9],24))
    # 4
    print(elevator([2,3],10))
    # -1

골드바흐의 추측(Goldbach's conjecture)은 오래전부터 알려진 정수론의 미해결 문제로, 2보다 큰 모든 짝수는 두 개의 소수(Prime number)의 합으로 표시할 수 있다는 것이다. 이때 하나의 소수를 두 번 사용하는 것은 허용한다. - 위키백과

위 설명에서 2보다 큰 모든 짝수를 두 소수의 합으로 나타낸 것을 골드바흐 파티션이라고 합니다.

예) 100 == 47 + 53 56 == 19 + 37

2보다 큰 짝수 n이 주어졌을 때, 골드바흐 파티션을 출력하는 코드를 작성하세요.

개념

1. n이 주어졌을 때 n까지의 모든 소수 출력
2. (n - 출력된 소수)가 소수인지 확인
3. 오름차순 정렬
4. 중복제거 예를들어 n = 100 이면 [47, 53], [53, 47]은 중복됨

코드

import math

def goldbach(n):
    if n <= 2 or n % 2 != 0:
        return '적절하지 못한 입력(2보다 큰 짝수)'

    def prime(n):
        array = [True for i in range(n+1)]
        # 에라토스테네스의 체
        for i in range(2, int(math.sqrt(n)) + 1):
            if array[i] == True:
                j = 2
                while i * j <= n:
                    array[i * j] = False
                    j += 1
        return [ i for i in range(2, n+1) if array[i] ]

    result = []

    prime_arr = prime(n) # n까지 전체 소수 리스트
    for i in prime_arr:
        if (n - i) in prime_arr: # n-소수가 소수인지 판별
            values = sorted([i, n-i])
            if values in result: # 중복 제거
                return result
            result.append(values)

    return result

if __name__ == '__main__':

    print(goldbach(100))
    # [[3, 97], [11, 89], [17, 83], [29, 71], [41, 59], [47, 53]]
    print(goldbach(56))
    # [[3, 53], [13, 43], [19, 37]]

행렬 2개가 주어졌을 때 곱할 수 있는 행렬인지 확인하고 곱할 수 있다면 그 결과를, 곱할 수 없다면 -1을 출력하는 프로그램을 만들어주세요.

input = [[1, 2],[2, 4]],[[1, 0],[0, 3]]
output = [[1,6], [2,12]]

### 개념

1. 행렬 치환
2. 곱의 합
3. 연산

### 코드
```python
def matrix_mul(matrix_1, matrix_2):
    #곱의 합
    def sum_product(arr1, arr2):
        return sum([a * b for a, b in zip(arr1, arr2)])

    #행렬 뒤집기
    def tran(matrix):           
        return [[matrix[j][i] for j in range(len(matrix))] for i in range(len(matrix[0]))]

    return [[sum_product(arr1, arr2) for arr2 in tran(matrix_2)] for arr1 in matrix_1]

if __name__ == '__main__':
    print(matrix_mul([[1, 2],[2, 4]],[[1, 0],[0, 3]]))
    # [[1,6], [2,12]]

Reference

제주 코딩 베이스 캠프 코딩 페스티벌 python 100제

1. 기준이되는 값에서 나머지의 값의 절대차를 구함
2. 오름차순 정렬 후 가장 첫번째 값 추출

코드

def near_country(countries, name):
    base = countries[name]
    del countries[name]

    for k, v in countries.items():
        countries[k] = abs(countries[k] - base)

    sorted_contries = sorted(countries.items(), key = lambda item: item[1])

    print(a[0][0], a[0][1], sep=' ')


if __name__ == '__main__':
    nationWidth = {
        'korea': 220877,
        'Rusia': 17098242,
        'China': 9596961,
        'France': 543965,
        'Japan': 377915,
        'England' : 242900 }

    near_country(nationWidth, 'korea')

    # England 22023

Reference

제주 코딩 베이스 캠프 코딩 페스티벌 python 100제

퀵 정렬(Quicksort)은 찰스 앤터니 리처드 호어가 개발한 정렬 알고리즘이다.

다른 원소와의 비교만으로 정렬을 수행하는 비교 정렬에 속한다.

퀵 정렬은 n개의 데이터를 정렬할 때, 최악의 경우에는 O(n2)번의 비교를 수행하고, 평균적으로 O(n log n)번의 비교를 수행한다.

개념

1. 두 개의 배열을 생성(front, back)
2. 중간 값(pivot) 보다 크면 뒤로 작으면 앞으로 이동
3. 더이상 중간 값이 없을 때 까지 1~2번 반복(재귀함수 사용)

코드

import collections

def quick_sort(arr):
    length = len(arr)

    if length <= 1:
      return arr

    mid = arr.pop(length//2)
    front = []
    back = []

    for i in arr:
      if i < mid:
        front.append(i)
      else:
        back.append(i)

      return quick_sort(front) + [mid] + quick_sort(back)

print(quick_sort([4,3,2,1]))
# [1, 2, 3, 4]

Reference

제주 코딩 베이스 캠프 코딩 페스티벌 python 100제

1. 처음에 모든 원판은 A기둥에 꽂혀 있다.
2. 모든 원판의 지름은 다르다.
3. 이 원반은 세 개의 기둥 중 하나에 반드시 꽂혀야 한다.
4. 작은 원반 위에 큰 원반을 놓을 수 없다.
5. 한 번에 하나의 원판(가장 위에 있는 원판) 만을 옮길 수 있다.

코드

def hanoi(n, from_obj, to_obj, aux_obj):
    if n == 1:
        print(f'{from_obj}->{to_obj}')
        return
    hanoi(n-1, from_obj, aux_obj, to_obj)
    print(f'{from_obj}->{to_obj}')
    hanoi(n-1, aux_obj, to_obj, from_obj)

if __name__ == '__main__':
    print('n=======>1')
    hanoi(1, 'A', 'B', 'C')
    print('n=======>2')
    hanoi(2, 'A', 'B', 'C')

결과

» python3 playground.py
n=======>1
A->B
n=======>2
A->C
A->B
C->B
n=======>3
A->B
A->C
B->C
A->B
C->A
C->B
A->B

Reference

제주 코딩 베이스 캠프 코딩 페스티벌 python 100제

unittest의 setUp처럼 테스트 실행 DB 등록같이 전처리가 필요할 때 사용한다.

특히 @pytest.fixture(scope=??) scope에 따라 실행되는 범위가 다르기 때문에 직접 실행하면서 이해해보자

@pytest.fixture()

import pytest

@pytest.fixture(scope='function')
def fixture():
    print('\nTest_1 Fixture!!')

def test_a(fixture):
    print('\ntest_a')

def test_b(fixture):
    print('\ntest_b')

def test_c(fixture):
    print('\ntest_c')

# in test_2.py
from test_1 import fixture 

def test_2_a(fixture):
    print('\ntest_a')

def test_2_d(fixture):
    print('\ntest_b')

def test_2_c(fixture):
    print('\ntest_c')

아무 scope=function일 때는 모든 test 함수에서 실행된다.

@pytest.fixture(scope='module')

# in test_1.py
import pytest

@pytest.fixture(scope='module')
def fixture():
    print('\nTest_1 Fixture!!')

def test_a(fixture):
    print('\ntest_a')

def test_b(fixture):
    print('\ntest_b')

def test_c(fixture):
    print('\ntest_c')

# in test_2.py
from test_1 import fixture 

def test_2_a(fixture):
    print('\ntest_a')

def test_2_d(fixture):
    print('\ntest_b')

def test_2_c(fixture):
    print('\ntest_c')

다음과 같이 파일을 하나 더생성한뒤 fixture함수를 import 하였다.

scope를 module로 하면 fixture가 파일 단위로 실행됨을 알 수 있다.

 gunicorn --bind 0.0.0.0:8000 app:app

이렇게 해서 사용할 port를 지정해준다.

안된다.. 아무래도 나의 앱에서는 Aplication Factory를 사용해서 그런것같다.

app.py에서 create_app으로 app을 만들어 주기 때문에 다음과 같이 사용하자.

 gunicorn --bind 0.0.0.0:8000 app:create_app()

안되는데?

당황하지말고 구글링을 좀 더한뒤 다음과 같이 해보자 ㅎㅎ

 gunicorn --bind 0.0.0.0:8000 "app:create_app()"

잘된다.

통신도 원활히 되는 것을 확인할 수 있다.

good~

목표

flask로 user CR 만들기

• 준비물
  • flask
  • mysql
  • pytest
  • sqlalchemy

    하나. 시작은 항상 가상환경부터

    sudo apt-get install python3-venv #설치
    python3 -m venv venv #생성
    . venv/bin/activate #활성화

    이쯤되면 외울때도 됫겠다.

둘. requirments 설치

pip install flask
pip install sqlalchemy
pip install mysqlclient

셋. app 작성

# in main.py
from flask import Flask, jsonify, request
from flask.views import MethodView

from sqlalchemy    import create_engine, text

import config

def create_app(test_config=None):
    app = Flask(__name__)

    if test_config:
        app.config.update(test_config)
    database = create_engine(app.config['DB_URL'], encoding = 'utf-8', max_overflow = 0)

    # 뷰
    class UserListView(MethodView):
        def __init__(self, database):
            self.db = database

        def get(self):
            users = self.db.execute("SELECT * FROM users")
            return jsonify({'users': [dict(row) for row in users] }), 200

    # 라우팅
    app.add_url_rule('/user', view_func=UserListView.as_view('user',db))
    return app

if __name__ == '__main__':
    app = create_app()
    app.run(host='0.0.0.0', debug=True, port=8000)

간단하게 sqlachemy를 이용해서 유저정보를 작성하는 앱을 만들었다.

여기서 테스트 시 test_config를 통해 DB_URL를 TEST용 DB로 바꿔 줄 것이다.

넷. test 작성

테스트를 작성하기 위해서는 다음을 고려해야한다.

1. 테스트용 DB를 따로 사용해야한다.
2. 테스트가 돌때마다 테스트용 DB의 테이블이 초기화되어야한다.

• 테스트용 DB 적용하기

# in test_main.py
...

import pytest
from main import create_app
from sqlalchemy import create_engine, text
import config

TEST_CONFIG = {
    'TESTING': True,
    'DB_URL': config.TEST_DB_URL
}

@pytest.fixture(scope='session') # 테스트 실행시 한번만 실행
def app():
    app = create_app(TEST_CONFIG)
    return app

@pytest.fixture(scope='session')
def db():
    db = create_engine(TEST_CONFIG['DB_URL'], encoding = 'utf-8', max_overflow = 0)
    return db

@pytest.fixture # 매 테스트 실행 마다 실행
def client(app, db):
    db_file(db, 'test.sql')
    client = app.test_client()
    return client

...

여기서 create_app에 설정해둔 TEST_CONFIG를 넣어주면서 테스트용 DB에서 테스트가 진행될 수 있도록 바꾸었다.

또한 db_file이라는 파일이라는 함수를 작성하여 'test.sql' 정보를 읽어 테스트 함수마다 초기화되도록 fixture function 단위인 client에 넣어주었다.

• @pytest.fixture 함수형 테스트에서 사전작업이 필요한 부분들을 묶어 줄 수 있다. 아무 설정도 하지않으면 매 테스트 마다 실행된다.

• @pytest.fixture(scope='session') 테스트 전반에 한번만 수행된다. app을 생성하거나 db를 지정하는 동작은 테스트시 한번만 하면되므로 session으로 지정하였다.

test 파일에 다음 항목을 추가해주었다.

다섯. sql 파일 만들기

UserView를 테스트 하기 위해서 users 테이블를 만들고 한명의 유저를 Insert하는 sql를 넣었다.

-- in test.sql
DROP TABLE users;
CREATE TABLE users
(
    `id`             INT             NOT NULL    AUTO_INCREMENT COMMENT 'pk',
    `name`           VARCHAR(100)    NOT NULL    COMMENT '유저이름',
    PRIMARY KEY (id)
);
INSERT INTO users (id, name) VALUES (1, 'user1');

• 테스트 중간중간 자신의 일이 잘진행되고 있는지 확인하는것은 시간을 매우 아껴준다.

  #mysql 접속
  >> source test.sql
  >> desc users

  

다섯. test 작성 및 실행

이제 실제 UserView를 테스트하는 함수를 작성해보자.

# in test_main.py
...
def test_get_user(client):
    response = client.get('/user')
    assert response.status_code == 200
    assert response.json == { 'users': [{'id': 1, 'name':'user1'}] } 

이렇게 하면 test.sql에서 만들어진 user1의 유저를 가져온다.

작성할 것은 다 작성했다. 실행해보자.

pytest -s

테스트가 잘 통과되었다.

DB에 user1이 잘들어왔는지도 확인해보자.

이런 방식으로 테스트DB를 원하는대로 세팅하고 테스트를 진행하면된다.

기타

• pytest는 test_* 파일을 자동으로 읽어온다. 테스트파일은 반드시 test_로 시작하도록 한다.
• fuction도 동일하게 def test_로 시작하도록 한다.

전체 파일

main.py

# in main.py
from flask import Flask, jsonify, request
from flask.views import MethodView

from sqlalchemy    import create_engine, text

import config

def create_app(test_config=None):
    app = Flask(__name__)

    app.config.from_pyfile("config.py")

    if test_config:
        app.config.update(test_config)

    db = create_engine(app.config['DB_URL'], encoding = 'utf-8', max_overflow = 0)

    # 뷰
    class UserListView(MethodView):
        def __init__(self, database):
            self.db = database

        def get(self):
            users = self.db.execute("SELECT * FROM users")
            return jsonify({'users': [dict(row) for row in users] }), 200

    # 라우팅
    app.add_url_rule('/user', view_func=UserListView.as_view('user',db))
    return app

if __name__ == '__main__':
    app = create_app()
    app.run(host='0.0.0.0', debug=True, port=8000)

tast_main.py

# in test_main.py
import pytest
from main import create_app
from sqlalchemy import create_engine, text
import config

TEST_CONFIG = {
    'TESTING': True,
    'DB_URL': config.TEST_DB_URL
}

def db_file(db, filename):
    sql_lines = []
    with open(filename, 'r') as file_data:
        # .sql를 주석을 제외하고 라인별로 분류
        sql_lines = [line.strip('\n') for line in file_data if not line.startswith('--') and line.strip('\n')]

        with db.connect() as conn:
            sql_command = ''
            for line in sql_lines:
                sql_command += line
                # ; 나오면 execute
                if sql_command.endswith(';'):
                    try:
                        conn.execute(text(sql_command))
                    except Exception as e:
                        print('Fail DB Reset!!')
                        print(e)
                        return False
                    finally:
                        sql_command = ''
    return True

@pytest.fixture(scope='session') # 테스트 실행시 한번만 실행
def app():
    app = create_app(TEST_CONFIG)
    return app

@pytest.fixture(scope='session')
def db():
    db = create_engine(TEST_CONFIG['DB_URL'], encoding = 'utf-8', max_overflow = 0)
    return db

@pytest.fixture # 매 테스트 실행 마다 실행
def client(app, db):
    db_file(db, 'test.sql')
    client = app.test_client()
    return client

def test_get_user(client):
    response = client.get('/user')
    assert response.status_code == 200
    assert response.json == { 'users': [{'id': 1, 'name':'user1'}] } 

config.py

DB = {
    'host': 'localhost',
    'user': 'root',
    'pass': '',
    'db': 'my_flask'
}

DB_URL = f'mysql://{DB["user"]}:{DB["pass"]}@{DB["host"]}/{DB["db"]}?charset=utf8'

TEST_DB = 'test'

TEST_DB_URL = f'mysql://{DB["user"]}:{DB["pass"]}@{DB["host"]}/{TEST_DB}?charset=utf8'

test.sql

DROP TABLE users;
CREATE TABLE users
(
    `id`             INT             NOT NULL    AUTO_INCREMENT COMMENT 'pk',
    `name`           VARCHAR(100)    NOT NULL    COMMENT '유저이름',
    PRIMARY KEY (id)
);
INSERT INTO users (id, name) VALUES (1, 'user1');

이런 코드가 있다.

from flask import Flask, jsonify, request
app = Flask(__name__)

@app.route('/a', methods=['POST'])
def a():
    try:
        data = request.get_json()['a']
        return jsonify({'message': data}), 200
    except KeyError:
        return jsonify({'massage': 'KEY_ERROR'}), 400

@app.route('/b', methods=['POST'])
def b():
    try:
        data = request.get_json()['b']
        return jsonify({'message': data}), 200
    except KeyError:
        return jsonify({'massage': 'KEY_ERROR'}), 400

if __name__ == '__main__':
    app.run(host='127.0.0.1', debug=True, port=8001)

분명 잘동작하는 코드지만 KeyError 관련해서 중복의 냄새가 난다.

특히 POST의 경우 KeyError처리는 필수적이기 때문에 이러한 중복은 계속늘어날 수 있다.

flask에서는 친절하게도 이런 에러처리를 도와주는 decorator가 있다. 도와줘요 공식문서

대충 decorator에 처리하고 싶은 에러를 넣어주면 해당함수에서 처리해주고 decorator를 쓰기 싫으면 app에다 등록도 할 수 있다는 것 같다.

손가락이 근질근질하다. 바로 해보자.

적용하기

from flask import Flask, jsonify, request
app = Flask(__name__)

# error handler
@app.errorhandler(KeyError)
def handler_key_error():
    return jsonify({'massage': 'KEY_ERROR'}), 400

@app.route('/a', methods=['POST'])
def a():
#    try:
    data = request.get_json()['a']
    return jsonify({'message': data}), 200
#    except KeyError:
#        return jsonify({'massage': 'KEY_ERROR'}), 400

@app.route('/b', methods=['POST'])
def b():
#    try:
        data = request.get_json()['b']
        return jsonify({'message': data}), 200
#    except KeyError:
#        return jsonify({'massage': 'KEY_ERROR'}), 400

if __name__ == '__main__':
    app.run(host='127.0.0.1', debug=True, port=8001)

결과

기능은 동일하지만 중복은 줄었다.

배웠으면 당장 refactoring하자

• 설치

  # in 가상환경
  pip install pymysql

• 작성

  # in connection.py
  import pymysql
  

def get_connect(): return pymysql.connect( host='127.0.0.1', password='', user='root', db='test', chaeset='utf8' )

user 생성까지 한번 해보자

def create_user(name): conn = get_connection() curs = conn.cursor() ok = curs.execute("INSERT INTO users(name) VALUES (%s)", name) conn.commit() conn.close()

return json.dumps({'rows': ok})

- 결과 
python shell로 확인 해보자
```python
from connect import *

print(create_user('user'))
{"rows": 1}