일 | 월 | 화 | 수 | 목 | 금 | 토 |
---|---|---|---|---|---|---|
1 | 2 | 3 | 4 | 5 | ||
6 | 7 | 8 | 9 | 10 | 11 | 12 |
13 | 14 | 15 | 16 | 17 | 18 | 19 |
20 | 21 | 22 | 23 | 24 | 25 | 26 |
27 | 28 | 29 | 30 |
- sshpass
- nohup
- centos
- appium
- insert
- Jupyter
- ubuntu
- STF
- perfect
- create table
- 실행권한
- PYTHON
- nmap
- openpyxl
- mysql
- STF_PortForwarding
- nGrinder
- Jupyter Notebook
- kitura
- SWIFT
- Materials
- postgresql
- GoCD
- port forwarding
- 28015
- postgres
- appium server
- ftp
- ssh
- rethinkdb
- Today
- Total
don't stop believing
Flask로 API 서버 만들기 (5) - User Operations 본문
Developing API Sample Server by Flask
Original Post: How to structure a Flask-RESTPlus web service for production builds
Github: https://github.com/cosmic-byte/flask-restplus-boilerplate
Flask로 API 서버 만들기 (1) - 개발 환경 준비
Flask로 API 서버 만들기 (2) - config 와 실행 확인
Flask로 API 서버 만들기 (3) - User 테이블 만들기
Flask로 API 서버 만들기 (4) - Testing
Flask로 API 서버 만들기 (5) - User Operations
Flask로 API 서버 만들기 (6) - Security and Authentication
User 테이블을 만들었으니 User 처리에 대한 기능을 만들어 보겠습니다.
만들 기능은 아래와 같습니다.
1. 새로운 User 생성
2. 등록된 User의 public_id 확인하기
3. 등록된 모든 User 불러오기
service 폴더 안에 user_service.py 파일을 만들어 줍니다.
1$ sudo vim ./app/main/service/user_service.py
user_service.py 파일안에 아래 script를 추가합니다.
123456789101112131415161718192021222324252627282930313233343536373839404142import uuidimport datetimefrom app.main import dbfrom app.main.model.user import Userdef save_new_user(data):user = User.query.filter_by(email=data['email']).first()if not user:new_user = User(public_id=str(uuid.uuid4()),email=data['email'],username=data['username'],password=data['password'],registered_on=datetime.datetime.utcnow())save_changes(new_user)response_object = {'status': 'success','message': 'Successfully registered.'}return response_object, 201else:response_object = {'status': 'fail','message': 'User already exists. Please Log in.',}return response_object, 409def get_all_users():return User.query.all()def get_a_user(public_id):return User.query.filter_by(public_id=public_id).first()def save_changes(data):db.session.add(data)db.session.commit()
save_new_user 함수는 신규 유저를 추가하는 기능입니다.
먼저 email이 같은 유저가 있는지를 확인합니다. 없다면 새로운 User를 추가합니다.
get_all_users 함수는 모든 User 정보를 불러오는 기능입니다.
get_a_user 함수는 public_id를 조회하고 매칭되는 User 정보를 불어옵니다.
save_changes 함수는 User 데이터를 저장(업데이트)하는 함수입니다.
이번에는 Utility 기능들을 만들어 보겠습니다.
main 폴더 아래 util 폴더를 만들고 dto.py 파일을 만들어 줍니다.
dto는 data transfer object로 데이터를 전달하는 기능을 합니다.
12$ mkdir ./app/main/util$ sudo vim ./app/main/util/dto.py
dto.py에 User 데이터를 전달하는 함수를 만들어 줍니다.
1234567891011from flask_restplus import Namespace, fieldsclass UserDto:api = Namespace('user', description='user related operations')user = api.model('user', {'email': fields.String(required=True, description='user email address'),'username': fields.String(required=True, description='user username'),'password': fields.String(required=True, description='user password'),'public_id': fields.String(description='user Identifier')})
user라는 Namespace를 만들고 api 변수에 할당합니다.
그리고 api.model을 이용해 user 데이터를 담을 필드를 만들어 줍니다.
이런게 하면 Flask 안에서 User 데이터를 담에 전달할 수 있습니다.
이제 User 데이터를 http로 처리할 controller를 만들어 주겠습니다.
controller 폴더 안에 user_controller.py 파일을 만들어 줍니다.
1$ sudo vim ./app/main/controller/user_controller.py
user_controller.py 파일안에 아래 script를 추가합니다.
12345678910111213141516171819202122232425262728293031323334353637383940from flask import requestfrom flask_restplus import Resourcefrom ..util.dto import UserDtofrom ..service.user_service import save_new_user, get_all_users, get_a_userapi = UserDto.api_user = UserDto.user@api.route('/')class UserList(Resource):@api.doc('list_of_registered_users')@api.marshal_list_with(_user, envelope='data')def get(self):"""List all registered users"""return get_all_users()@api.response(201, 'User successfully created.')@api.doc('create a new user')@api.expect(_user, validate=True)def post(self):"""Creates a new User """data = request.jsonreturn save_new_user(data=data)@api.route('/')@api.param('public_id', 'The User identifier')@api.response(404, 'User not found.')class User(Resource):@api.doc('get a user')@api.marshal_with(_user)def get(self, public_id):"""get a user given its identifier"""user = get_a_user(public_id)if not user:api.abort(404)else:return user
app 폴더 밑에있는 __init__.py 파일안에 API Server의 entry point를 잡아줍니다.
1$ sudo vim ./app/__init__.py
__init__.py파일 안에 아래와 같이 추가합니다.
12345678910111213141516# app/__init__.pyfrom flask_restplus import Apifrom flask import Blueprintfrom .main.controller.user_controller import api as user_nsblueprint = Blueprint('api', __name__)api = Api(blueprint,title='FLASK RESTPLUS API BOILER-PLATE WITH JWT',version='1.0',description='a boilerplate for flask restplus web service')api.add_namespace(user_ns, path='/user')
manage.py 파일 안에 blueprint를 추가해 줍니다.
1$ sudo vim ./manage.py
추가할 내용은 blueprint 모듈 추가(from app import blueprint)와 app.register_blueprint(blueprint) 입니다.
123456789101112131415161718192021222324252627282930313233343536import osimport unittestfrom flask_migrate import Migrate, MigrateCommandfrom flask_script import Managerfrom app import blueprintfrom app.main import create_app, dbfrom app.main.model import userapp = create_app(os.getenv('BOILERPLATE_ENV') or 'dev')app.register_blueprint(blueprint)app.app_context().push()manager = Manager(app)migrate = Migrate(app, db)manager.add_command('db', MigrateCommand)@manager.commanddef run():app.run()@manager.commanddef test():"""Runs the unit tests."""tests = unittest.TestLoader().discover('app/test', pattern='test*.py')result = unittest.TextTestRunner(verbosity=2).run(tests)if result.wasSuccessful():return 0return 1if __name__ == '__main__':manager.run()
이제 manage.py로 API 서버를 실행합니다.
12345678$ python manage.py run* Serving Flask app "app.main" (lazy loading)* Environment: development* Debug mode: on* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)* Restarting with stat* Debugger is active!* Debugger PIN: 145-484-411
브라우저를 열고 http://localhost:5000으로 접속합니다.
만약 Flask Server가 원격으로 있다면 외부에서 접근을 허용해야 합니다.
manage.py 안에 app.run()을 호출할때 host='0.0.0.0'을 추가합니다.
app.run(host='0.0.0.0')
그리고 다시 실행하면 Running on 에 아래와 같이 나오게 되며 외부에서도 접속이 가능해 집니다.
* Running on http://0.0.0.0:5000/ (Press CTRL+C to quit)
12345678$ python manage.py run* Serving Flask app "app.main" (lazy loading)* Environment: development* Debug mode: on* Running on http://0.0.0.0:5000/ (Press CTRL+C to quit)* Restarting with stat* Debugger is active!* Debugger PIN: 145-484-411
다시 브라우저를 열고 내부 IP와 port 5000으로 접속합니다.
POST /user 텝을 열고 Try it out 버튼을 클릭한 후 아래와 같이 email, username, password, public_id을 json 타입에 맞게 입력하고 Execute를 클릭합니다.
그럼 아래와 같이 결과가 나옵니다.
정상적으로 등록되었는지 GET /user/를 클릭해 보겠습니다.
GET /user 텝을 열고 Try it out 버튼을 클릭하고 바로 Execute 버튼을 클릭합니다.
그럼 등록한 tongchun의 정보가 보입니다.
여기까지 User 테이블을 만들고 User_Coltroller와 blueprint를 이용해 swagger를 실행해 데이터를 입력하고 출력하는 것을 확인했습니다.
'Python > Flask' 카테고리의 다른 글
Flask로 API 서버 만들기 (7) - Route protection and Authorization (1) | 2018.11.18 |
---|---|
Flask로 API 서버 만들기 (6) - Security and Authentication (0) | 2018.11.12 |
Flask로 API 서버 만들기 (4) - Testing (0) | 2018.11.12 |
Flask로 API 서버 만들기 (3) - User 테이블 만들기 (2) | 2018.11.12 |
Flask로 API 서버 만들기 (2) - config 와 실행 확인 (2) | 2018.11.07 |