I am creating a simple app using python/flask with mariadb
I am almost finishing my CRUD, however, the update is not working.
The code is:
from flask import Flask, request
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
import pymysql
pymysql.install_as_MySQLdb()
app = Flask(__name__)
db = SQLAlchemy(app)
app.config["SQLALCHEMY_DATABASE_URI"] = "mysql://{username}:{password}@{server}/{db}".format(username = 'root', password = '', server = 'localhost', db = 'baby-tracker')
class Event(db.Model):
id = db.Column(db.Integer,primary_key=True)
description = db.Column(db.String(100),nullable=False)
def __repr__(self):
return f"Event: {self.description}"
def __init__(self,description):
self.description = description
def format_event(event):
return {
"description": event.description,
"id": event.id
}
#Main page
@app.route('/')
def hello():
return 'Hello mate!'
#Create an event
@app.route('/events', methods=["POST"])
def create_event():
description = request.get_json('description')
event = Event(**description)
db.session.add(event)
db.session.commit()
return format_event(event)
#Get all events
@app.route('/events', methods=["GET"])
def get_events():
events = Event.query.order_by(Event.id.asc()).all()
event_list = []
for event in events:
event_list.append(format_event(event))
return {"events": event_list}
#Get a single event
@app.route('/events/<id>', methods=["GET"])
def get_event(id):
#Could do this: if request.method == 'GET':
event = Event.query.filter_by(id=id).one()
formatted_event = format_event(event)
return formatted_event
#Get a single event
@app.route('/events/<id>', methods=["DELETE"])
def delete_event(id):
event = Event.query.filter_by(id=id).one()
db.session.delete(event)
db.session.commit()
return f'Event id {id} deleted'
#Update event
@app.route('/events/<id>', methods=["PUT"])
def update_event(id):
event = Event.query.filter_by(id=id)
description = request.get_json('description')
event.update(dict(description = description))
db.session.commit()
formatted_event = format_event(event)
return formatted_event
if __name__ == '__main__':
app.run()
I am getting the following error for the PUT request:
JSON I am sending in the put request using postman:
{ "description": "new test" }
The error I get:
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\flask\app.py", line 2091, in __call__
return self.wsgi_app(environ, start_response)
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\flask\app.py", line 2076, in wsgi_app
response = self.handle_exception(e)
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\flask\app.py", line 2073, in wsgi_app
response = self.full_dispatch_request()
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\flask\app.py", line 1518, in full_dispatch_request
rv = self.handle_user_exception(e)
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\flask\app.py", line 1516, in full_dispatch_request
rv = self.dispatch_request()
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\flask\app.py", line 1502, in dispatch_request
return self.ensure_sync(self.view_functions[rule.endpoint])(**req.view_args)
File "C:\Users\PC\Documents\estudo_python\projeto_integrado\backend\app.py", line 77, in update_event
event.update(dict(description = description))
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\sqlalchemy\orm\query.py", line 3300, in update
execution_options={"synchronize_session": synchronize_session},
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\sqlalchemy\orm\session.py", line 1712, in execute
result = conn._execute_20(statement, params or {}, execution_options)
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\sqlalchemy\engine\base.py", line 1705, in _execute_20
return meth(self, args_10style, kwargs_10style, execution_options)
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\sqlalchemy\sql\elements.py", line 334, in _execute_on_connection
self, multiparams, params, execution_options
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\sqlalchemy\engine\base.py", line 1582, in _execute_clauseelement
cache_hit=cache_hit,
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\sqlalchemy\engine\base.py", line 1944, in _execute_context
e, statement, parameters, cursor, context
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\sqlalchemy\engine\base.py", line 2125, in _handle_dbapi_exception sqlalchemy_exception, with_traceback=exc_info[2], from_=e
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\sqlalchemy\util\compat.py", line 208, in raise_
raise exception
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\sqlalchemy\engine\base.py", line 1901, in _execute_context
cursor, statement, parameters, context
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\sqlalchemy\engine\default.py", line 736, in do_execute
cursor.execute(statement, parameters)
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\pymysql\cursors.py", line 148, in execute
result = self._query(query)
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\pymysql\cursors.py", line 310, in _query
conn.query(q)
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\pymysql\connections.py", line 548, in query
self._affected_rows = self._read_query_result(unbuffered=unbuffered)
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\pymysql\connections.py", line 775, in _read_query_result
result.read()
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\pymysql\connections.py", line 1156, in read
first_packet = self.connection._read_packet()
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\pymysql\connections.py", line 725, in _read_packet
packet.raise_for_error()
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\pymysql\protocol.py", line 221, in raise_for_error
err.raise_mysql_exception(self._data)
File "C:\Users\PC\.virtualenvs\backend-_y8cE5wK\lib\site-packages\pymysql\err.py", line 143, in raise_mysql_exception
raise errorclass(errno, errval)
sqlalchemy.exc.ProgrammingError: (pymysql.err.ProgrammingError) (1064, 'You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near \'\'description\': "\'new test\'"} WHERE event.id = \'1\'\' at line 1')
[SQL: UPDATE event SET description=%(description)s WHERE event.id = %(id_1)s]
[parameters: {'description': {'description': 'new test'}, 'id_1': '1'}]
(Background on this error at: https://sqlalche.me/e/14/f405)
I was getting I similar error with the POST event, but I solved it unpacking the dictionary with '**' tryed something similar with the PUT but haven´t get a positive result.
The update works now (updated code below), but I´ve got a new error. I do not understand why the one() function is not working since the row that was updated is there.
from flask import Flask, request
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
import pymysql
from flask_cors import CORS
import json
pymysql.install_as_MySQLdb()
app = Flask(__name__)
db = SQLAlchemy(app)
CORS(app)
app.config["SQLALCHEMY_DATABASE_URI"] = "mysql://{username}:{password}@{server}/{db}".format(username = 'root', password = '', server = 'localhost', db = 'baby-tracker')
class Event(db.Model):
id = db.Column(db.Integer,primary_key=True)
description = db.Column(db.String(100),nullable=False)
created_at = db.Column(db.DateTime,nullable=False,default=datetime.utcnow)
updated_at = db.Column(db.DateTime,nullable=True,default=datetime.utcnow)
def __repr__(self):
return f"Event: {self.description}"
def __init__(self,description):
self.description = description
def format_event(event):
return {
"description": event.description,
"id": event.id,
"created_at": event.created_at,
"updated_at": event.updated_at
}
#Main page
@app.route('/')
def hello():
return 'Hello mate!'
#Create an event
@app.route('/events', methods=["POST"])
def create_event():
description = request.get_json('description')
event = Event(**description)
db.session.add(event)
db.session.commit()
return format_event(event)
#Get all events
@app.route('/events', methods=["GET"])
def get_events():
events = Event.query.order_by(Event.id.asc()).all()
event_list = []
for event in events:
event_list.append(format_event(event))
return {"events": event_list}
#Get a single event
@app.route('/events/<id>', methods=["GET"])
def get_event(id):
#Could do this: if request.method == 'GET':
event = Event.query.filter_by(id=id).one()
formatted_event = format_event(event)
return formatted_event
#Get a single event
@app.route('/events/<id>', methods=["DELETE"])
def delete_event(id):
event = Event.query.filter_by(id=id).one()
db.session.delete(event)
db.session.commit()
return f'Event id {id} deleted'
#Update event
@app.route('/events/<id>', methods=["PUT"])
def update_event(id):
event = Event.query.filter_by(id=id)
description = request.get_json('description')['description']
#print(description['description'])
#print(json.dumps(description))
event.update(dict(description = description))
db.session.commit()
formatted_event = format_event(event.one())
return formatted_event
if __name__ == '__main__':
app.run()