Flask REST API

  from flask import Blueprint, request, jsonify, abort
from app.models import Post
from app import db

api = Blueprint('api', __name__)

@api.route('/posts/<int:id>', methods=['GET'])
def get_post(id):
    post = Post.query.get_or_404(id)
    return jsonify({
        'id': post.id,
        'title': post.title,
        'body': post.body,
        'created_at': post.created_at.isoformat(),
    })

@api.route('/posts/<int:id>', methods=['PUT'])
def update_post(id):
    post = Post.query.get_or_404(id)
    data = request.get_json()

    if not data or 'title' not in data:
        abort(400, description='Title is required')

    post.title = data['title']
    post.body = data.get('body', post.body)
    db.session.commit()
    return jsonify({'message': 'Updated'})

@api.route('/posts/<int:id>', methods=['DELETE'])
def delete_post(id):
    post = Post.query.get_or_404(id)
    db.session.delete(post)
    db.session.commit()
    return '', 204
  

  pip install flask-marshmallow marshmallow-sqlalchemy
  

  from flask_marshmallow import Marshmallow
from marshmallow import fields, validate

ma = Marshmallow()

class PostSchema(ma.SQLAlchemySchema):
    class Meta:
        model = Post

    title = fields.String(required=True, validate=validate.Length(min=1, max=200))
    body = fields.String(required=True)

post_schema = PostSchema()
posts_schema = PostSchema(many=True)

@api.route('/posts', methods=['POST'])
def create_post():
    data = request.get_json()
    errors = post_schema.validate(data)
    if errors:
        return jsonify(errors), 400

    post = Post(title=data['title'], body=data['body'], user_id=1)
    db.session.add(post)
    db.session.commit()
    return post_schema.jsonify(post), 201
  

  pip install flask-jwt-extended
  

  from flask_jwt_extended import (
    JWTManager, create_access_token, jwt_required, get_jwt_identity
)

jwt = JWTManager()

def create_app():
    app = Flask(__name__)
    app.config['JWT_SECRET_KEY'] = 'your-secret-key'
    jwt.init_app(app)
    ...

@api.route('/login', methods=['POST'])
def login():
    data = request.get_json()
    user = User.query.filter_by(username=data['username']).first()
    if user and user.check_password(data['password']):
        token = create_access_token(identity=user.id)
        return jsonify(access_token=token)
    return jsonify({'error': 'Invalid credentials'}), 401

@api.route('/protected', methods=['GET'])
@jwt_required()
def protected():
    user_id = get_jwt_identity()
    return jsonify(logged_in_as=user_id)
  

  @api.errorhandler(404)
def not_found(error):
    return jsonify({'error': 'Resource not found'}), 404

@api.errorhandler(400)
def bad_request(error):
    return jsonify({'error': str(error.description)}), 400
  

  import pytest
from app import create_app, db

@pytest.fixture
def client():
    app = create_app()
    app.config['TESTING'] = True
    app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'

    with app.test_client() as client:
        with app.app_context():
            db.create_all()
        yield client

def test_create_post(client):
    response = client.post('/api/posts', json={
        'title': 'Test', 'body': 'Content'
    })
    assert response.status_code == 201