728x90

5.Flask sqlalchemy - get, create

controller 부터 service 그리고 dao를 작성하여 get과 create를 구현 해보겠습니다.

https://github.com/kschoi93/flask-example

Controller 작성

example_create_model = example.model('create', {
    'name': fields.String
})

@example.route('/name/<name>')
class ExampleRouteGetId(Resource):
    def get(self, name: str) -> json:
        result = example_services.example_route_get_id(name)
        return {'message':'성공',
                'result':result}, 200

@example.route('/create')
@example.param('name', 'Create User')
class ExampleRouteCreate(Resource):
    @example.expect(example_create_model, validate=True)
    @example.marshal_with(example_create_model)
    def post(self) -> json:
        result = example_services.example_route_create(request.data)
        return {'message':'성공'}, 200'

ExampleRouteGetId에 대한 api 문서

Id를 가져오는데, get 요청 시 path에 있는 name으로 데이터를 받아오도록 서비스로 넘겨주는 역할을 합니다.

@example.route('/name/<name>')
class ExampleRouteGetId(Resource):
    def get(self, name: str) -> json:
        result = example_services.example_route_get_id(name)
        return {'message':'성공',
                'result':result}, 200

ExampleRouteCreate에 대한 api 문서

post로 받은 데이터를 유요한 데이터인지 확인하고 서비스로 넘기도록 합니다.

@example.route('/create')
@example.param('name', 'Create User')
class ExampleRouteCreate(Resource):
    @example.expect(example_create_model, validate=True)
    @example.marshal_with(example_create_model)
    def post(self) -> json:
        result = example_services.example_route_create(request.data)
        return {'message':'성공'}, 200'

except와 marshal_with에 사용되는 모델

example_create_model = example.model('create', {
    'name': fields.String
})

@example.expect(example_create_model, validate=True)
@example.marshal_with(example_create_model)

Service 작성

dao 객체를 생성하고 해당 dao 객체에서 get과 create를 하도록 작성합니다.

def example_route_get_id(name: str) -> str:
    dao = ExampleDAO()
    return dao.get(name=name)

def example_route_create(data: bytes):
    create_data = json.loads(data)

    name: str = create_data['name']
    dao = ExampleDAO()
    result: int = dao.create(name=name)

Dao 작성

각각 get과 create를 작성합니다.

class ExampleDAO(object):
    def __init__(self):
        pass

    def get(self, name: str) -> str:
        user = ExampleUser.query.filter_by(name=name).first()
        return user.name

    def create(self, name: str):
        user = ExampleUser(name=name)
        db.session.add(user)
        db.session.commit()

Create 테스트

다음과 같이 http://127.0.0.1:5000/api/example/create 경로로 body에 json을 실어 보냅니다.

database를 확인하면 다음과 같이 admin이 생성되어 있습니다

get 테스트

admin이 있는지 path param으로 실어 보내 테스트하면 다음과 같은 결과가 나타납니다.