개발일기

Rekognition이란 aws 객체 탐지 ai api다.

from datetime import datetime
from http import HTTPStatus
from os import access
from flask import request
from flask_jwt_extended import create_access_token, get_jwt, get_jwt_identity, jwt_required
from flask_restful import Resource
from mysql.connector.errors import Error
from config import Config
import boto3

boto3와 라이브러리를 import한다.

class ObjectDetctionResource(Resource):

    def get(self):

        # 1. 클라이언트로부터 데이터를 받아온다.

        filename = request.args['filename']

        # 2. 위의 파일은, S3에 저장되어 있어야 한다.
        # rekognition 을 이용해서, object detection 한다.

        client = boto3.client('rekognition', 'ap-northeast-2', aws_access_key_id = Config.ACCESS_KEY, aws_secret_access_key = Config.SECRET_ACCESS)
        response = client.detect_labels(Image = {'S3Object': {'Bucket':Config.S3_BUCKET, 'Name':filename}}, MaxLabels = 10)

        for label in response['Labels']:
            print ("Label: " + label['Name'])
            print ("Confidence: " + str(label['Confidence']))
            print ("Instances:")
            for instance in label['Instances']:
                print ("  Bounding box")
                print ("    Top: " + str(instance['BoundingBox']['Top']))
                print ("    Left: " + str(instance['BoundingBox']['Left']))
                print ("    Width: " +  str(instance['BoundingBox']['Width']))
                print ("    Height: " +  str(instance['BoundingBox']['Height']))
                print ("  Confidence: " + str(instance['Confidence']))
                print()

            print ("Parents:")
            for parent in label['Parents']:
                print ("   " + parent['Name'])
            print ("----------")
            print ()
        
        return {'result':'success', 'Labels':response['Labels']}, 200

클라이언트로부터 파일이름을 받고, boto3.client로 연결한다. detect_labels의 파라미터에 s3와 파일이름을 넣고 라벨 수를 조정해준 뒤 결과를 받아온다.

from datetime import datetime
from http import HTTPStatus
from os import access
from flask import request
from flask_jwt_extended import create_access_token, get_jwt, get_jwt_identity, jwt_required
from flask_restful import Resource
from mysql.connector.errors import Error
from mysql_connection import get_connection
import mysql.connector
from config import Config
import boto3

pip install boto3로 설치하고 import 한다. boto3는 aws를 파이썬에서 연결할 수 있도록 해주는 라이브러리다.

class PostingResource(Resource):
    def post(self):

        # 1. 클라이언트로부터 데이터 받아온다.
        # photo(file), content(text)

        if 'photo' not in request.files:
            return {'error', '파일을 업로드하세요'},  400

        file = request.files['photo']
        content = request.form['content']

        # 2. S3에 파일 업로드
        # 파일명을 우리가 변경해 준다.
        # 파일명은, 유니크하게 만들어야 한다.
        current_time = datetime.now()
        new_file_name = current_time.isoformat().replace(':', '_') + '.jpg'

        # 유저가 올린 파일의 이름을, 내가 만든 파일명으로 변경
        file.filename = new_file_name

        # S3 에 업로드 하면 된다.
        # AWS 의 라이브러리를 사용해야 한다.
        # 이 파이썬 라이브러리를 boto3 라이브러리다!
        # boto3 라이브러리 설치
        # pip install boto3

        s3 = boto3.client('s3', aws_access_key_id = Config.ACCESS_KEY, aws_secret_access_key = Config.SECRET_ACCESS)

        try:
            s3.upload_fileobj(file, Config.S3_BUCKET, file.filename, ExtraArgs = {'ACL':'public-read', 'ContentType':file.content_type})

        except Exception as e:
            return {'error':str(e)}, 500

        # 3. DB에 저장

        try :
            # 데이터 insert 
            # 1. DB에 연결
            connection = get_connection()

            # 2. 쿼리문 만들기
            query = '''insert into posting
                    (content, imgUrl)
                    values
                    ( %s, %s);'''
            
            record = (content, new_file_name)

            # 3. 커서를 가져온다.
            cursor = connection.cursor()

            # 4. 쿼리문을 커서를 이용해서 실행한다.
            cursor.execute(query, record)

            # 5. 커넥션을 커밋해줘야 한다 => 디비에 영구적으로 반영하라는 뜻
            connection.commit()

            # 6. 자원 해제
            cursor.close()
            connection.close()

        except mysql.connector.Error as e :
            print(e)
            cursor.close()
            connection.close()
            return {"error" : str(e)}, 503

        return {'result':'success'}

포스트맨에 url 입력하고 경로 연결해준다음 

클라이언트로부터 데이터를 받아온다.

boto3.client로 연결하고 upload_fileobj로 파일 업로드를 한다.

mysql db에는 글 내용과 파일이름을 업로드한다.

읽어올때는 s3주소뒤에 파일이름을 붙이면 파일을 읽어올 수 있다.

from flask import Flask
from flask_jwt_extended import JWTManager
from flask_restful import Api
from config import Config

from resources.recipe import RecipeListResource
from resources.recipe_info import RecipeResource
from resources.recipe_publish import RecipePublishResource
from resources.user import UserLoginResource, UserLogoutResource, UserRegisterResource, jwt_blacklist


app = Flask(__name__)

# 환경변수 셋팅
app.config.from_object(Config)

# JWT 토큰 라이브러리 만들기
jwt = JWTManager(app)

# 로그아웃 된 토큰이 들어있는 set을, jwt에 알려준다.
@jwt.token_in_blocklist_loader
def check_if_token_is_revoked(jwt_header, jwt_payload):
    jti = jwt_payload['jti']
    return jti in jwt_blacklist



api = Api(app)

# 경로와 리소스(API 코드)를 연결한다.
api.add_resource(RecipeListResource, '/recipes')
api.add_resource(RecipeResource, '/recipes/<int:recipe_id>')
api.add_resource(RecipePublishResource, '/recipes/<int:recipe_id>/publish')
api.add_resource(UserRegisterResource, '/users/register')
api.add_resource(UserLoginResource, '/users/login')
api.add_resource(UserLogoutResource, '/users/logout')

if __name__ == "__main__":
    app.run()

@jwt.token_in_blocklist_loader를 적고 함수를 만든다.

로그아웃된 토큰이 들어있는 set을, jwt에 알려준다.

# 로그아웃 기능을 하는 클래스
class UserLogoutResource(Resource):

    @jwt_required()
    def post(self):

        jti = get_jwt()['jti']
        print(jti)

        jwt_blacklist.add(jti)

        return {'result', 'success'}, 200

로그아웃 기능을 하는 클래스다. @jwt_required를 적어주고 jti를 만들고 집합에 넣어준다.

config.py 파일

class Config:
    JWT_SECRET_KEY = 'yhacademy1029##hello'
    JWT_ACCESS_TOKEN_EXPIRES = True
    PROPAGATE_EXCEPTIONS = True

user.py 파일

access_token = create_access_token(user_id, expires_delta = datetime.timedelta(minutes = 1))

create_access_token의 expires_delta 파라미터에 datatime.timedelta를 대입한다.

위의 예제에서는 유효기간이 1분이다

    @jwt_required()
    def put(self, recipe_id):

        # body에서 전달된 데이터를 처리
        data = request.get_json()

        user_id = get_jwt_identity()

        # 디비 업데이트 실행코드
        try:
            # 데이터 업데이트
            # 1. DB에 연결
            connection = get_connection()

            ### 먼저 recipe_id에 들어있는 user_id가
            ### 이사람인지 먼저 확인한다.

            query = '''select user_id
                        from recipe
                        where id = %s;'''
            record = (recipe_id, )

            cursor = connection.cursor(dictionary = True)

            cursor.execute(query, record)

            result_list = cursor.fetchall()

            print(result_list)

            if len(result_list) == 0:
                cursor.close()
                connection.close()
                return {'error': '레시피 아이디가 잘못되었습니다'}, 400
                
            recipe = result_list[0]

            if recipe['user_id'] != user_id:
                cursor.close()
                connection.close()
                return {'error':'남의 레시피를 수정할 수 없습니다.'}, 401


            # 2. 쿼리문 만들기
            query = '''update recipe
                    set name = %s, description = %s,
                    cook_time = %s,
                    directions = %s
                    where id = %s;'''

            record = (data['name'], data['description'], data['cook_time'], data['directions'], recipe_id)
            
            # 3. 커서를 가져온다.
            cursor = connection.cursor()

            # 4. 쿼리문을 커서를 이용해서 실행한다.
            cursor.execute(query, record)

            # 5. 커넥션을 커밋해줘야 한다 => 디비에 영구적으로 반영하라는 뜻
            connection.commit()

            # 6. 자원 해제
            cursor.close()
            connection.close()

        except mysql.connector.Error as e:
            print(e)
            cursor.close()
            connection.close()
            return {'error':str(e)}, 503
        
        return {'result':'sucess'}, 200

@jwt_required()를 함수 앞에 써주고 get_jwt_identity()를 통해 토큰을 받는다.

포스트맨에서 테스트해보려면

Header KEY에 Authoriztion을 넣고 VALUE에 Bearer 토큰 을 넣은다음 테스트하면 된다.

전체코드

import datetime
from http import HTTPStatus
from flask import request
from flask_jwt_extended import create_access_token
from flask_restful import Resource
from mysql.connector.errors import Error
from mysql_connection import get_connection
import mysql.connector
from email_validator import validate_email, EmailNotValidError

from utils import check_password, hash_password

class UserRegisterResource(Resource):
    def post(self):
        # {
        #     "username":"홍길동"
        #     "email": "abc@naver.com",
        #     "password": "1234"
        # } 

        # 1. 클라이언트가 body에 보내준 json을 받아온다.
        data = request.get_json()

        # 2. 이메일 주소형식이 제대로 된 주소형식인지 
        # 확인하는 코드 작성.
        try:
            validate_email(data['email'])
        except EmailNotValidError as e:
            return {"error" : str(e)}, 400

        # 3. 비밀번호의 길이가 유효한지 체크한다.
        # 비번길이는 4자리 이상, 12자리 이하로만!
        if len(data['password']) < 4 or len(data['password']) > 12:
            return {'error':'비번길이확인하세요'}, 400

        # 4. 비밀번호를 암호화 한다.
        hashed_password = hash_password(data['password'])

        # 5. 데이터베이스에 회원정보를 저장한다!!

        try:
            # 데이터 업데이트
            # 1. DB에 연결
            connection = get_connection()

            # 2. 쿼리문 만들기
            query = '''insert into user
                    (username, email, password)
                    values
                    (%s, %s, %s);'''

            record = (data['username'], data['email'], hashed_password )
            
            # 3. 커서를 가져온다.
            cursor = connection.cursor()

            # 4. 쿼리문을 커서를 이용해서 실행한다.
            cursor.execute(query, record)

            # 5. 커넥션을 커밋해줘야 한다 => 디비에 영구적으로 반영하라는 뜻
            connection.commit()

            # 5-1. 디비에 저장된 아이디값 가져오기
            user_id = cursor.lastrowid

            # 6. 자원 해제
            cursor.close()
            connection.close()

        except mysql.connector.Error as e:
            print(e)
            cursor.close()
            connection.close()
            return {'error':str(e)}, 503

        # user_id 를 바로 보내면 안되고, 
        # JWT로 암호화 해서 보내준다.
        # 암호화 하는 방법
        access_token = create_access_token(user_id, expires_delta = datetime.timedelta(minutes = 1))

        return {'result': 'success', 'access_token': access_token}, 200

class UserLoginResource(Resource):
    def post(self):
        # {
        #     "email": "abc@naver.com",
        #     "password": "1234"
        # } 

        # 1. 클라이언트가 body에 보내준 json을 받아온다.
        data = request.get_json()

        try:
            # 1. DB에 연결
            connection = get_connection()

            # 2. 쿼리문 만들기
            query = '''select *
                    from user
                    where email = (%s);'''

            record = (data['email'], )
            
            # 3. 커서를 가져온다.
            cursor = connection.cursor()

            # 4. 쿼리문을 커서를 이용해서 실행한다.
            cursor = connection.cursor(dictionary = True)
            cursor.execute(query, record)

            # 4-1 select id를 가져온다
            result_list = cursor.fetchall()

        
            i = 0
            for record in result_list :
                result_list[i]['created_at'] = record['created_at'].isoformat()
                result_list[i]['updated_at'] = record['updated_at'].isoformat()
                i = i + 1                

            # 5. 커넥션을 커밋해줘야 한다 => 디비에 영구적으로 반영하라는 뜻
            connection.commit()

            # 6. 자원 해제
            cursor.close()
            connection.close()

            # result_list의 행의 갯수가 1개이면,
            # 유저 데이터를 정상적으로 받아온것이고,
            # 행의 갯수가 0이면, 요청한 이메일은, 회원가입이
            # 되어 있지 않은 이메일이다.

            if len(result_list) != 1:
                return {'error': '회원가입이 안된 이메일입니다.'}, 400

            # 4. 비밀번호가 맞는지 확인한다.
            user_info = result_list[0]

            # data['password'] 와 user_info['password']를 비교
            check = check_password(data['password'], user_info['password'])
            
            if check == False:
                return {'error': '비밀번호가 맞지 않습니다.'}, 400

        except mysql.connector.Error as e:
            print(e)
            cursor.close()
            connection.close()
            return {'error':str(e)}, 503

        access_token = create_access_token(user_info['id'],  expires_delta = datetime.timedelta(minutes = 1))

        return {'result': 'success', 'access_token ': access_token }, 200

user_id를 바로 보내지 않고, JWT로 암호화 해서 보내준다.

라이브러리 설치 pip install flask-jwt-extended로 설치한다.

config.py 파일을 만든다.

class Config:
    JWT_SECRET_KEY = 'yhacademy1029##hello'
    JWT_ACCESS_TOKEN_EXPIRES = True
    PROPAGATE_EXCEPTIONS = True

키를 입력하고, 설정을 한다.

app.py 파일에 다음과 같은 코드를 추가한다.

from config import Config
app = Flask(__name__)

# 환경변수 셋팅
app.config.from_object(Config)

# JWT 토큰 라이브러리 만들기
jwt = JWTManager(app)

환경변수 셋팅을 하고 jwt 토큰 라이브러리를 만든다.

Postman 이란 (https://www.getpostman.com/)

A powerful GUI platform to make your API development faster & easier, from building API requests through testing, documentation and sharing.

postman은 api개발을 빠르고 쉽게 만들어 주는 gui플랫폼이다. 

포스트맨 좌측에서 collection을 만든다.

콜렉션의 더보기를 클릭해서 add request를 누른다.

post ,put같이 body가 필요할때 json 형식으로 대입할 수 있다. http method도 url왼쪽에서 고를 수 있다.

params에서 파라미터들을 입력하면 query string에 알맞은 형식으로 들어간다.

send 를 누르면 api테스트가 되고 아래에서 return 결과를 받아볼 수 있다.

app.py파일이 실행파일이고 resources폴더안에 recipe.py와 recipe_info.py파일이 들어있다.

mysql_connection.py파일에는 mysql에 연결하기위한 코드들이 작성되어 있다.

app.py 코드

from flask import Flask
from flask_restful import Api

from resources.recipe import RecipeListResource
from resources.recipe_info import RecipeResource


app = Flask(__name__)

api = Api(app)

# 경로와 리소스(API 코드)를 연결한다.
api.add_resource(RecipeListResource, '/recipes')
api.add_resource(RecipeResource, '/recipes/<int:recipe_id>')

if __name__ == "__main__":
    app.run()

플라스크를 import하고 기본 구조 작성뒤 경로와 리소스를 연결한다.

recipe.py 코드

from http import HTTPStatus
from flask import request
from flask_restful import Resource
from mysql.connector.errors import Error
from mysql_connection import get_connection
import mysql.connector

### API 를 만들기 위한 클래스 작성
### class(클래스) 란??  변수와 함수로 구성된 묶음!
### 클래스는 상속이 가능하다!
### API를 만들기 위한 클래스는, flask_restful 라이브러리의
### Resource 클래스를 상속해서 만들어야 한다.

class RecipeListResource(Resource):
    # restful api 의 method 에 해당하는 함수 작성
    def post(self) :
        # api 실행 코드를 여기에 작성

        # 클라이언트에서, body 부분에 작성한 json 을
        # 받아오는 코드
        data = request.get_json()

        # 받아온 데이터를 디비 저장하면 된다.
        try :
            # 데이터 insert 
            # 1. DB에 연결
            connection = get_connection()

            # 2. 쿼리문 만들기
            query = '''insert into recipe
                    (name, description, cook_time, directions)
                    values
                    ( %s , %s , %s ,%s);'''
            
            record = (data['name'], data['description'], data['cook_time'], data['directions']  )

            # 3. 커서를 가져온다.
            cursor = connection.cursor()

            # 4. 쿼리문을 커서를 이용해서 실행한다.
            cursor.execute(query, record)

            # 5. 커넥션을 커밋해줘야 한다 => 디비에 영구적으로 반영하라는 뜻
            connection.commit()

            # 6. 자원 해제
            cursor.close()
            connection.close()

        except mysql.connector.Error as e :
            print(e)
            cursor.close()
            connection.close()
            return {"error" : str(e)}, 503

        return {"result" : "success"}, 200

    def get(self) :
        # 쿼리 스트링으로 오는 데이터는 아래처럼 처리해준다.
        offset = request.args.get('offset')
        limit = request.args.get('limit')

        # 디비로부터 데이터를 받아서, 클라이언트에 보내준다.
        try :
            connection = get_connection()

            query = '''select *
                    from recipe
                    limit '''+offset+''' , '''+limit+''';'''
            
            # select 문은, dictionary = True 를 해준다.
            cursor = connection.cursor(dictionary = True)

            cursor.execute(query)

            # select 문은, 아래 함수를 이용해서, 데이터를 가져온다.
            result_list = cursor.fetchall()

            print(result_list)

            # 중요! 디비에서 가져온 timestamp 는 
            # 파이썬의 datetime 으로 자동 변경된다.
            # 문제는! 이데이터를 json 으로 바로 보낼수 없으므로,
            # 문자열로 바꿔서 다시 저장해서 보낸다.
            i = 0
            for record in result_list :
                result_list[i]['created_at'] = record['created_at'].isoformat()
                result_list[i]['updated_at'] = record['updated_at'].isoformat()
                i = i + 1                

            cursor.close()
            connection.close()

        except mysql.connector.Error as e :
            print(e)
            cursor.close()
            connection.close()

            return {"error" : str(e)}, 503


        return { "result" : "success" , 
                "count" : len(result_list) ,
                "result_list" : result_list }, 200

insert와 select가 post, get 메소드로 구현되어있다. 

recipe_info.py 코드

from http import HTTPStatus
from flask import request
from flask_restful import Resource
from mysql.connector.errors import Error
from mysql_connection import get_connection
import mysql.connector

class RecipeResource(Resource):
    # 클라이언트로부터 /recipes/3 이런식으로 경로를 처리하므로
    # 숫자는 바뀌므로, 변수로 처리해준다.
    def get(self, recipe_id):

        # 디비에서, recipe_id에 들어있는 값에 해당되는
        # 데이터를 select 해온다.
        try :
            connection = get_connection()

            query = '''select *
                    from recipe
                    where id = %s;'''

            record = (recipe_id, )
            
            # select 문은, dictionary = True 를 해준다.
            cursor = connection.cursor(dictionary = True)

            cursor.execute(query, record)

            # select 문은, 아래 함수를 이용해서, 데이터를 가져온다.
            result_list = cursor.fetchall()

            print(result_list)

            # 중요! 디비에서 가져온 timestamp 는 
            # 파이썬의 datetime 으로 자동 변경된다.
            # 문제는! 이데이터를 json 으로 바로 보낼수 없으므로,
            # 문자열로 바꿔서 다시 저장해서 보낸다.
            i = 0
            for record in result_list :
                result_list[i]['created_at'] = record['created_at'].isoformat()
                result_list[i]['updated_at'] = record['updated_at'].isoformat()
                i = i + 1                

            cursor.close()
            connection.close()

        except mysql.connector.Error as e :
            print(e)
            cursor.close()
            connection.close()

            return {"error" : str(e)}, 503 

        return {'result':'success',
                'info':result_list[0]}

    # 데이터를 업데이트하는 API들은 put 함수를 사용한다.
    def put(self, recipe_id):

        # body에서 전달된 데이터를 처리
        data = request.get_json()

        # 디비 업데이트 실행코드
        try:
            # 데이터 업데이트
            # 1. DB에 연결
            connection = get_connection()

            # 2. 쿼리문 만들기
            query = '''update recipe
                    set name = %s, description = %s,
                    cook_time = %s,
                    directions = %s
                    where id = %s;'''

            record = (data['name'], data['description'], data['cook_time'], data['directions'], recipe_id)
            
            # 3. 커서를 가져온다.
            cursor = connection.cursor()

            # 4. 쿼리문을 커서를 이용해서 실행한다.
            cursor.execute(query, record)

            # 5. 커넥션을 커밋해줘야 한다 => 디비에 영구적으로 반영하라는 뜻
            connection.commit()

            # 6. 자원 해제
            cursor.close()
            connection.close()

        except mysql.connector.Error as e:
            print(e)
            cursor.close()
            connection.close()
            return {'error':str(e)}, 503
        
        return {'result':'sucess'}, 200

    # 삭제하는 delete 함수
    def delete(self, recipe_id):
        try:
            # 데이터 삭제
            # 1. DB에 연결
            connection = get_connection()

            # 2. 쿼리문 만들기
            query = '''delete from recipe
                    where id = %s;'''

            record = (recipe_id, )
            
            # 3. 커서를 가져온다.
            cursor = connection.cursor()

            # 4. 쿼리문을 커서를 이용해서 실행한다.
            cursor.execute(query, record)

            # 5. 커넥션을 커밋해줘야 한다 => 디비에 영구적으로 반영하라는 뜻
            connection.commit()

            # 6. 자원 해제
            cursor.close()
            connection.close()

        except mysql.connector.Error as e:
            print(e)
            cursor.close()
            connection.close()
            return {'error':str(e)}, 503

        return {'result':'success'}, 200

select, update, delete가 get, put, delete 메소드로 구현되어 있다.

API(Application Programming Interface)란?

응용 프로그램에서 사용할 수 있도록, 운영 체제나 프로그래밍 언어가 제공하는 기능을 제어할 수 있게 만든 인터페이스를 뜻한다. API를 통해 소스 및 DB에는 접근하지 못하게 하고 해당 프로그램을 사용할 수 있도록 기능을 제공하는 것이다

REST(Representational State Transfer)란?

서버나 서비스에 존재하는 모든 자원(이미지, 동영상, DB자원)에 고유한 URL를 부여해 활용하는 것 = 자원을 정의하고 자원에 대한 주소를 지정하는 방법론을 의미한다.