Введение:
Python Flask — это популярная веб-инфраструктура, которая позволяет быстро и легко создавать веб-приложения, включая REST API. В этой статье мы рассмотрим процесс создания RESTful API с использованием Python Flask. К концу этого руководства вы будете иметь общее представление о том, как создавать конечные точки, обрабатывать методы HTTP и возвращать ответы JSON.
Шаг 1: Настройка приложения Flask:
Чтобы начать работу с приложением Flask, выполните следующие действия:
1.1. Установите Flask и необходимые зависимости:
$ pip install flask flask_sqlalchemy flask_jwt_extended
1.2. Создайте новый каталог для вашего проекта и перейдите к нему:
$ mkdir todo-api $ cd todo-api
1.3. Создайте новый файл Python с именем app.py
и откройте его в предпочитаемом вами редакторе кода.
1.4. Импортируйте необходимые модули и инициализируйте Flask:
from flask import Flask from flask_sqlalchemy import SQLAlchemy from flask_jwt_extended import JWTManager app = Flask(__name__) basedir = pathlib.Path(__file__).parent.resolve() app.config['SQLALCHEMY_DATABASE_URI'] = f"sqlite:///{basedir / 'todo.db'}" # Modify the database URI as per your choice app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False app.config['SECRET_KEY'] = 'your-secret-key' # Modify with your own secret key. For example: b'_5#y2L"F4Q8z\n\xec]/' db = SQLAlchemy(app) jwt = JWTManager(app)
1.5. Определите базовый маршрут, чтобы убедиться, что Flask работает правильно:
@app.route('/') def index(): return 'Welcome to the To-Do API!'
1.6. Запустите приложение Flask:
if __name__ == '__main__': app.run(debug=True)
Шаг 2. Интеграция с базой данных:
В этом разделе мы интегрируем базу данных в наше приложение Flask с помощью SQLAlchemy.
2.1. Определите модели базы данных для пользователей и элементов списка дел. Откройте app.py
и добавьте следующий код:
from datetime import datetime class User(db.Model): __tablename__ = 'user' id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(50), unique=True, nullable=False) password = db.Column(db.String(255), nullable=False) class TodoItem(db.Model): __tablename__ = 'todos' id = db.Column(db.Integer, primary_key=True) title = db.Column(db.String(100), nullable=False) description = db.Column(db.Text, nullable=True) completed = db.Column(db.Boolean, default=False) created_at = db.Column(db.DateTime, default=datetime.utcnow) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
2.2. Создайте таблицы базы данных. Откройте терминал и выполните следующие команды:
$ python >>> from app import app, db >>> app.app_context().push() >>> db.create_all() >>> exit()
Шаг 3. Создание конечных точек регистрации и аутентификации пользователей:
Далее мы реализуем конечные точки регистрации и аутентификации пользователей.
3.1. Импортируйте необходимые модули и добавьте следующие маршруты в app.py
:
from flask import request, jsonify from werkzeug.security import generate_password_hash, check_password_hash from flask_jwt_extended import create_access_token @app.route('/register', methods=['POST']) def register(): data = request.get_json() hashed_password = generate_password_hash(data.get('password')) new_user = User(username=data.get('username'), password=hashed_password) db.session.add(new_user) db.session.commit() return jsonify({'message': 'User registered successfully'}) @app.route('/login', methods=['POST']) def login(): data = request.get_json() user = User.query.filter_by(username=data.get('username')).first() if not user or not check_password_hash(user.password, data.get('password')): return jsonify({'message': 'Invalid username or password'}), 401 # Generate a JWT token access_token = create_access_token(identity=user.id) return jsonify({'access_token': access_token})
Шаг 4. Реализация функциональности списка дел:
В этом разделе мы реализуем основные функции управления элементами списка дел.
4.1. Добавьте следующий маршрут к app.py
для создания новых элементов списка дел:
from flask_jwt_extended import get_jwt_identity, , jwt_required @app.route('/todos', methods=['POST']) @jwt_required(fresh=False) def create_todo(): data = request.get_json() title = data.get('title') description = data.get('description') new_todo = TodoItem(title=title, description=description, user_id=get_jwt_identity()) db.session.add(new_todo) db.session.commit() return jsonify({'message': 'To-Do item created successfully'})
4.2. Добавьте следующий маршрут к app.py
для получения всех элементов списка дел для аутентифицированного пользователя:
@app.route('/todos', methods=['GET']) @jwt_required(fresh=False) def get_todos(): user_id = get_jwt_identity() todos = TodoItem.query.filter_by(user_id=user_id).all() result = [] for todo in todos: result.append({ 'id': todo.id, 'title': todo.title, 'description': todo.description, 'completed': todo.completed, 'created_at': todo.created_at }) return jsonify(result)
4.3. Добавьте следующий маршрут к app.py
, чтобы обновить определенный элемент списка дел:
@app.route('/todos/<int:todo_id>', methods=['PUT']) @jwt_required(fresh=False) def update_todo(todo_id): todo = TodoItem.query.get(todo_id) if not todo or todo.user_id != get_jwt_identity(): return jsonify({'message': 'To-Do item not found'}), 404 data = request.get_json() todo.title = data.get('title') todo.description = data.get('description') todo.completed = data.get('completed') db.session.commit() return jsonify({'message': 'To-Do item updated successfully'})
4.4. Добавьте следующий маршрут к app.py
, чтобы удалить определенный элемент списка дел:
@app.route('/todos/<int:todo_id>', methods=['DELETE']) @jwt_required(fresh=False) def delete_todo(todo_id): todo = TodoItem.query.get(todo_id) if not todo or todo.user_id != get_jwt_identity(): return jsonify({'message': 'To-Do item not found'}), 404 db.session.delete(todo) db.session.commit() return jsonify({'message': 'To-Do item deleted successfully'})
Шаг 5. Добавление авторизации и аутентификации пользователя:
Чтобы добавить авторизацию и аутентификацию пользователей в наш To-Do API, мы будем использовать Flask-JWT-Extended.
5.1. Импортируйте необходимые модули и добавьте в app.py
следующий код:
from flask_jwt_extended import jwt_required, get_jwt_identity from datetime import timedelta # ... (Existing code) # Configure JWT settings app.config['JWT_SECRET_KEY'] = 'your-secret-key' # Modify with your own secret key app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(days=1) jwt = JWTManager(app)
5.2. Защитите соответствующие маршруты с помощью декоратора @jwt_required
, чтобы обеспечить аутентификацию:
@app.route('/todos', methods=['GET']) @jwt_required(fresh=False) def get_todos(): # Existing code... @app.route('/todos', methods=['POST']) @jwt_required(fresh=False) def create_todo(): # Existing code... @app.route('/todos/<int:todo_id>', methods=['PUT']) @jwt_required(fresh=False) def update_todo(todo_id): # Existing code... @app.route('/todos/<int:todo_id>', methods=['DELETE']) @jwt_required(fresh=False) def delete_todo(todo_id): # Existing code...
Шаг 6. Внедрение проверки и обработки ошибок:
Чтобы повысить надежность API, мы реализуем проверку и обработку ошибок.
6.1. Установите необходимые библиотеки:
$ pip install flask_marshmallow marshmallow
6. 2. Создайте новый файл с именем schemas.py
и добавьте следующий код:
from marshmallow import Schema, fields, validate class TodoSchema(Schema): id = fields.Int(dump_only=True) title = fields.Str(required=True, validate=validate.Length(max=100)) description = fields.Str(validate=validate.Length(max=500)) completed = fields.Bool() created_at = fields.DateTime(dump_only=True) todo_schema = TodoSchema() todos_schema = TodoSchema(many=True)
6.3. Обновите маршруты в app.py
для проверки входных данных и обработки ошибок:
from schemas import todo_schema, todos_schema # ... (Existing code) @app.route('/todos', methods=['POST']) @jwt_required def create_todo(): data = request.json errors = todo_schema.validate(data) if errors: return jsonify({'message': 'Validation errors', 'errors': errors}), 400 # Existing code... @app.route('/todos/<int:todo_id>', methods=['PUT']) @jwt_required def update_todo(todo_id): data = request.json errors = todo_schema.validate(data) if errors: return jsonify({'message': 'Validation errors', 'errors': errors}), 400 # Existing code...
Шаг 7. Расширенный функционал:
В этом разделе мы добавим расширенные функции в наш To-Do API, включая сортировку и фильтрацию, разбиение на страницы и пометку задач как выполненных.
7.1. Чтобы разрешить сортировку и фильтрацию элементов списка дел, мы улучшим конечную точку /todos
GET.
@app.route('/todos', methods=['GET']) @jwt_required def get_todos(): user_id = get_jwt_identity() # Get query parameters sort_by = request.args.get('sort_by', default='created_at') filter_completed = request.args.get('filter_completed') todos_query = TodoItem.query.filter_by(user_id=user_id) # Apply sorting if sort_by == 'title': todos_query = todos_query.order_by(TodoItem.title) elif sort_by == 'created_at': todos_query = todos_query.order_by(TodoItem.created_at.desc()) # Apply filtering if filter_completed: filter_completed = filter_completed.lower() == 'true' todos_query = todos_query.filter_by(completed=filter_completed) todos = todos_query.all() # Existing code...
7.2. Чтобы реализовать разбиение на страницы для конечной точки /todos
GET, мы изменим код следующим образом:
from flask import request @app.route('/todos', methods=['GET']) @jwt_required def get_todos(): user_id = get_jwt_identity() # Get query parameters page = request.args.get('page', default=1, type=int) per_page = request.args.get('per_page', default=10, type=int) todos_query = TodoItem.query.filter_by(user_id=user_id) # Apply pagination todos_query = todos_query.paginate(page=page, per_page=per_page) todos = todos_query.items # Existing code...
7.3. Чтобы добавить функцию пометки задач как выполненных, мы обновим конечную точку /todos/<int:todo_id>
PUT.
@app.route('/todos/<int:todo_id>', methods=['PUT']) @jwt_required def update_todo(todo_id): # Existing code... if 'completed' in data: todo.completed = data.completed # Existing code...
Шаг 8. Тестирование To-Do API:
Протестируйте конечные точки с помощью таких инструментов, как cURL или Postman.
8.1. Регистрация пользователя:
Чтобы зарегистрировать нового пользователя, установите запрос http
как POST
, установите URL-адрес как http://127.0.0.1:5000/register
, а затем в теле запроса отправьте username
и password
для нового пользователя.
Тело запроса должно иметь вид json
.
Если пользователь создан успешно, вы получите сообщение об успешном завершении, которое говорит User registered successfully
.
8.2. Логин пользователя:
Чтобы войти в систему пользователя, установите запрос http
как POST
, установите URL-адрес как http://127.0.0.1:5000/login
, а затем в теле запроса отправьте username
и password
для нового пользователя.
Тело запроса должно иметь вид json
.
Если username
и password
верны, вы получите сообщение, содержащее access_token
.
Что-то вроде:
8.3. Создайте новый To-Do в базе данных:
Чтобы создать новую задачу, установите запрос http
как POST
, установите URL-адрес как http://127.0.0.1:5000/todos
, а затем в теле запроса отправьте title
и description
для новой задачи.
Перед отправкой запроса обязательно отправьте access_token
вместе с запросом. access_token
можно установить в разделе Authorization
.
Установите тип авторизации как Bearer Token
и вставьте токен, который вы получили от входа в систему, в поле ввода Token
. Вот так:
Теперь отправьте запрос. Вы получите сообщение об успехе, которое говорит To-Do item created successfully
. Вот так:
8.4. Получить список элементов To-Do:
Чтобы создать новую задачу, установите запрос http
как GET
установите URL как http://127.0.0.1:5000/todos
без тела запроса.
Перед отправкой запроса убедитесь, что вы отправляете access_token
вместе с запросом.
Теперь отправьте запрос. Вы получите список всех задач, которые вы создали. Вот так:
8.5. Обновить элемент списка дел:
Чтобы создать новую задачу, установите запрос http
как PUT
, установите URL-адрес как http://127.0.0.1:5000/todos/<item_id>
, а затем в теле запроса отправьте title
, description
и завершено для задачи.
Перед отправкой запроса убедитесь, что вы отправляете access_token
вместе с запросом.
Теперь отправьте запрос. Вы получите сообщение об успешном завершении To-Do item updated successfully
Вот так:
8.6. Удалить элемент списка дел:
Чтобы создать новую задачу, установите http
запрос как DELETE
установите URL как http://127.0.0.1:5000/todos/<item_id>
. Тело запроса не требуется.
Перед отправкой запроса убедитесь, что вы отправляете access_token
вместе с запросом.
Теперь отправьте запрос. Вы получите сообщение об успехе, в котором говорится To-Do item deleted successfully
. Вот так:
В этом руководстве мы создали расширенный API REST To-Do с использованием Python Flask. Мы рассмотрели различные темы, такие как настройка приложения Flask, интеграция базы данных, реализация регистрации и аутентификации пользователей, а также создание основных функций управления элементами списка дел. Кроме того, мы изучили расширенные функции, такие как сортировка, фильтрация, нумерация страниц и пометка элементов как выполненных. Следуя этому руководству, вы теперь обладаете знаниями и инструментами для создания собственного надежного и многофункционального REST API с использованием Flask.
Не стесняйтесь исследовать дополнительные функции и улучшения для дальнейшей настройки и расширения функциональности вашего To-Do API.
Я надеюсь, что эта статья помогла вам понять процесс создания To-Do REST API с использованием Python Flask. Если у вас есть какие-либо вопросы или вам нужна дополнительная помощь, пожалуйста, не стесняйтесь спрашивать.
Удачного кодирования!