I want to implement a REST server with Flask in Python 3.7. In particular, I want to separate the controller (the one who handles the URLs) from the business logic because it seems more maintainable to me. Below is a sample code that represents my code (although I actually have it divided into different files and classes):
#!/usr/bin/env python3
import os
os.environ["APP_SETTINGS"] = "db.config.DevelopmentConfig"
os.environ["DATABASE_URL"] = "postgresql://user_db:12341234@localhost/user_db"
os.environ["FLASK_ENV"] = "development"
from flask import Flask, request, jsonify
from flask_cors import CORS
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
cors = CORS(app, resources={r"/api/*": {"origins": "*"}})
app.config.from_object(os.environ['APP_SETTINGS'])
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app) # to handle the database transactions
class Game(db.Model):
__tablename__ = "GAME"
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
name = db.Column(db.String(40), nullable=False)
platform = db.Column(db.String(40), nullable=False)
description = db.Column(db.String(255), nullable=True)
def __init__(self, name, platform, description):
self.name = name
self.platform = platform
self.description = description
def __repr__(self):
return f"<name {self.name}>"
class GameService:
def getById(self, id: int):
entities = Game.query.filter_by(id=id)
if len(entities) > 1:
raise Exception(f"There are multiple entities with the same ID: '{id}'")
return entities.first()
def getByName(self, name: str):
entities = Game.query.filter_by(name=name).first()
if len(entities) > 1:
raise Exception(f"There are multiple entities with the same NAME: '{name}'")
return entities.first()
def create(self, name: str, platform: str, description: str):
# Build entity to persist
new_entity = Game(name=name, platform=platform, description=description)
# Create entity in DB
db.session.add(new_entity)
db.session.commit()
return new_entity.id
def update(self, entity: Game):
# Update entity in DB
db.session.merge(entity)
db.session.commit()
return entity.id
def delete(self, entity_id: int):
entities = Game.query.filter_by(id=entity_id)
if len(entities) > 1:
raise Exception(f"There are multiple entities with the same ID: '{entity_id}'")
db.session.delete(entities.first())
db.session.commit()
class GameController:
def __init__(self):
self.service = GameService()
@app.route("/v1/game/create", methods=["GET", "POST"])
def createGame(self):
try:
# Create new game in DB
entity_id = self.service.create(name=request.args.get("name"),
platform=request.args.get("platform"),
description=request.args.get("description"),
)
# Build response
response = jsonify(status="SUCCESS")
response.headers.add("Access-Control-Allow-Origin", "*")
return response
except Exception as e:
response = jsonify(status="ERROR", error_code="001")
response.headers.add("Access-Control-Allow-Origin", "*")
return response
@app.route("/v1/game/delete", methods=["GET", "POST"])
def deleteGame(self):
try:
# Create new game in DB
self.service.delete(entity_id=int(request.args.get("id")))
# Build response
response = jsonify(status="SUCCESS")
response.headers.add("Access-Control-Allow-Origin", "*")
return response
except Exception as e:
response = jsonify(status="ERROR", error_code="002")
response.headers.add("Access-Control-Allow-Origin", "*")
return response
app.run(host="127.0.0.1", port=5000, debug=True)
The problem arises when I make, for example, the following call:
http://127.0.0.1:5000/v1/game/create?name=GTA&platform=PS4&description=shooting
The system returns the error:
TypeError: createGame() missing 1 required positional argument: 'self'
I am aware that if createGame is defined as @staticmethod (and, therefore, by removing the 'self' parameter), this call works properly, but not could use any class attribute within that method.
Does anyone know how I could fix this problem? That is, how could you define a flask route to work with a class method?
CodePudding user response：
I think flask's pluggable views should be able to solve your problem.
Another option would be the extension flask-classful. I don't have any experience with it though.
CodePudding user response：
Thanks for your answer @ffrosch.
I've used FlaskView as the documentation says: http://flask-classful.teracy.org/#customizing-the-route-base.
However, I have a new problem when I need to add a parameter in the controller's constructor. I was reading the doc from here but it does not work for me. I also was trying different options but without success.
I've updated my code in order to show my problem: now, GameService has a parameter in its constructor:
#!/usr/bin/env python3
import os
os.environ["APP_SETTINGS"] = "db.config.DevelopmentConfig"
os.environ["DATABASE_URL"] = "postgresql://user_db:12341234@localhost/user_db"
os.environ["FLASK_ENV"] = "development"
from flask import Flask, request, jsonify
from flask_classful import FlaskView, route
from flask_cors import CORS
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
cors = CORS(app, resources={r"/api/*": {"origins": "*"}})
app.config.from_object(os.environ["APP_SETTINGS"])
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app) # to handle the database transactions
class Game(db.Model):
__tablename__ = "GAME"
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
name = db.Column(db.String(40), nullable=False)
platform = db.Column(db.String(40), nullable=False)
description = db.Column(db.String(255), nullable=True)
def __init__(self, name, platform, description):
self.name = name
self.platform = platform
self.description = description
def __repr__(self):
return f"<name {self.name}>"
class GameService:
def __init__(self, config_file):
self.config_file = config_file
def getById(self, id: int):
entities = Game.query.filter_by(id=id)
if len(entities) > 1:
raise Exception(f"There are multiple entities with the same ID: '{id}'")
return entities.first()
def getByName(self, name: str):
entities = Game.query.filter_by(name=name).first()
if len(entities) > 1:
raise Exception(f"There are multiple entities with the same NAME: '{name}'")
return entities.first()
def create(self, name: str, platform: str, description: str):
# Build entity to persist
new_entity = Game(name=name, platform=platform, description=description)
# Create entity in DB
db.session.add(new_entity)
db.session.commit()
return new_entity.id
def update(self, entity: Game):
# Update entity in DB
db.session.merge(entity)
db.session.commit()
return entity.id
def delete(self, entity_id: int):
entities = Game.query.filter_by(id=entity_id)
if len(entities) > 1:
raise Exception(f"There are multiple entities with the same ID: '{entity_id}'")
db.session.delete(entities.first())
db.session.commit()
class GameController(FlaskView):
def __init__(self, config_file):
self.service = GameService(config_file)
@route("/create", methods=["GET", "POST"])
def createGame(self):
try:
# Create new game in DB
entity_id = self.service.create(name=request.args.get("name"),
platform=request.args.get("platform"),
description=request.args.get("description"),
)
# Build response
response = jsonify(status="SUCCESS")
response.headers.add("Access-Control-Allow-Origin", "*")
return response
except Exception as e:
response = jsonify(status="ERROR", error_code="001")
response.headers.add("Access-Control-Allow-Origin", "*")
return response
@route("/delete", methods=["GET", "POST"])
def deleteGame(self):
try:
# Create new game in DB
self.service.delete(entity_id=int(request.args.get("id")))
# Build response
response = jsonify(status="SUCCESS")
response.headers.add("Access-Control-Allow-Origin", "*")
return response
except Exception as e:
response = jsonify(status="ERROR", error_code="002")
response.headers.add("Access-Control-Allow-Origin", "*")
return response
# I want to use the following line:
# GameController.register(app, route_base="/v1/game", config_file="/path/to/config/file.txt")
# But my first try is:
GameController.register(app, "/path/to/config/file.txt")
app.run(host="127.0.0.1", port=5000, debug=True)