Flask Rest API - MantsSk/CA_PTUA14 GitHub Wiki

REST API kūrimo kodas

from flask import Flask, jsonify, request
from calendar import isleap

app = Flask(__name__)


@app.route("/", methods=['GET', 'POST'])
def index():
    if (request.method == 'POST'):
        some_json = request.get_json()
        return jsonify({'you sent': some_json})
    else:
        return jsonify({'about': 'Hello World'})


@app.route("/leap/<int:year>", methods=['GET'])
def keliamieji(year):
    if isleap(year):
        return jsonify({'result': "Leap"})
    else:
        return jsonify({'result': "Not leap"})


if __name__ == '__main__':
    app.run(port=8000, debug=True)

REST API kūrimo kodas (antras būdas panaudojant flask_restful)

pip3 install flask_restful
from flask import Flask, request
from flask_restful import Resource, Api
from calendar import isleap

app = Flask(__name__)
api = Api(app)


class HelloWorld(Resource):
    # Parametras 'self' yra būtinas
    # metodo pavadinimas turi būti norimo requesto pavadinimas: get, post, delete, etc.
    def get(self):
        return {'about': 'Hello world!'}

    # Parametras 'self' yra būtinas
    # metodo pavadinimas turi būti norimo requesto pavadinimas: get, post, delete, etc.
    def post(self):
        body = request.get_json()
        return {'you sent': body}, 200  # Galima nurodyti status kodą


class Leap(Resource):
    # Parametras 'self' yra būtinas
    # metodo pavadinimas turi būti norimo requesto pavadinimas: get, post, delete, etc.
    def get(self, year):
        if isleap(year):
            return {'result': "Leap"}
        else:
            return {'result': "Not leap"}


# užregistruojame endpointą, nurodome route'ą. kaip tą endpointą pasiekti
api.add_resource(HelloWorld, '/')
# užregistruojame endpointą, nurodome route'ą. kaip tą endpointą pasiekti
api.add_resource(Leap, '/leap/<int:year>')

if __name__ == '__main__':
    app.run(port=8000, debug=True)

Sukurto REST API panaudojimo kodas

pip3 install requests
import requests
import json

r = requests.get('http://127.0.0.1:8000/leap/2028')
dictionary = json.loads(r.text)
print(dictionary['result'])


import requests
import json

payload = {'name': 'Jonas', 'surname': 'Petras', 'age': 2000}
r = requests.post('http://127.0.0.1:8000/', json=payload)
dictionary = json.loads(r.text)
print(dictionary)

# {'you sent': {'name': 'Jonas', 'surname': 'Petras', 'age': 2000}}

REST API su duomenų baze kūrimas

pip3 install flask_marshmallow
pip3 install marshmallow-sqlalchemy
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from flask_migrate import Migrate
import os

basedir = os.path.abspath(os.path.dirname(__file__))

app = Flask(__name__)

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + \
    os.path.join(basedir, 'sqlite.db')

db = SQLAlchemy(app)
ma = Marshmallow(app)
Migrate(app, db)


class Task(db.Model):
    __tablename__ = 'tasks'
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column('title', db.String)
    done = db.Column('done', db.Boolean)


# Užduoties schema
class TaskSchema(ma.Schema):
    class Meta:
        fields = ('id', 'title', 'done')


task_schema = TaskSchema()
tasks_schema = TaskSchema(many=True)


# Crud
@app.route('/tasks/new', methods=['POST'])
def add_task():
    title = request.json['title']
    done = request.json['done']
    new_task = Task(title=title, done=done)
    db.session.add(new_task)
    db.session.commit()
    return task_schema.jsonify(new_task)


# cRud
@app.route('/tasks', methods=['GET'])
def all_tasks():
    all_tasks = Task.query.all()
    return tasks_schema.jsonify(all_tasks)


# cRud
@app.route('/tasks/<id>', methods=['GET'])
def get_task(id):
    task = Task.query.get(id)
    return task_schema.jsonify(task)


# crUd
@app.route('/tasks/<id>', methods=['PUT'])
def edit_task(id):
    task = Task.query.get(id)
    task.title = request.json['title']
    task.done = request.json['done']
    db.session.commit()
    return task_schema.jsonify(task)


# cruD
@app.route('/tasks/<id>', methods=['DELETE'])
def delete_task(id):
    task = Task.query.get(id)
    db.session.delete(task)
    db.session.commit()
    return task_schema.jsonify(task)


if __name__ == '__main__':
    app.run(host='127.0.0.1', port=8000, debug=True)


Traukiame duomenis iš API su DB

Įrašome užduotį:

nauja_uzduotis = {
    "title": "Išplauti indus",
    "done": False
}

r = requests.post('http://127.0.0.1:8000/tasks/new', json=new_task)
print(json.loads(r.text))

Nuskaitome visas užduotis:

r = requests.get('http://127.0.0.1:8000/tasks')
print(json.loads(r.text))

Nuskaitome vieną užduotį:

r = requests.get('http://127.0.0.1:8000/tasks/1')
print(json.loads(r.text))

Pakeičiame vieną užduotį:

nauja_uzduotis = {
    "title": "Išplauti grindis",
    "done": False
}

r = requests.put('http://127.0.0.1:8000/tasks/8', json=new_task)
print(json.loads(r.text))

Ištriname užduotį:

r = requests.delete('http://127.0.0.1:8000/tasks/7')
print(json.loads(r.text))

Užduotis

Sukurti Flask API programą, kuri:

Užduotis

Sukurti Flask API programą, kuri:

  • Per API (padavus JSON duomenis) leistų įvesti prekę į duomenų bazę (jos formavimui panaudoti sqlalchemy ORM).
  • Prekė turi turėti id, pavadinimą, kainą ir kiekį (galima keisti prekę į bet ką kitą, svarbu kad būtų viena lentelę ir turėtų kelis stulpelius)
  • Per API leistų sukurti prekę
  • Per API leistų peržiūrėti visas įvestas prekes
  • Per API leistų peržiūrėti vieną konkrečią prekę
  • Per API leistų pakoreguoti prekę
  • Per API leistų ištrinti prekę
  • Pasitestuoti kiekvieną endpointą (panaudoti bent po 1 kartą)
⚠️ **GitHub.com Fallback** ⚠️