I'm writing APIs in flask, and i'm asking myself if i use SQLAlchemy in a functional but wrong way.
Currently my app is like
# File app.__init__.py
db = SQLAlchemy()
app = Flask(__name__, instance_relative_config=False)
def create_app():
app.config.from_object("config.Config")
CORS(app)
db.init_app(app)
So i don't initialize any engine.
# File app.core.core.py
from flask import current_app
from app import db
...
# Blueprint Configuration
core_bp = Blueprint("core_bp", __name__)
# Swagger documentation configuration
authorizations = {"apikey": {"type": "apiKey", "in": "header", "name": "Authorization"}}
api = Api(core_bp, authorizations=authorizations)
ns = api.namespace("auth", description="User Authentification APIs")
...
@ns.route("/signup", methods=["POST"])
class SignupApi(Resource):
"""
Enpoint for sign up of users
"""
@api.expect(resource_fields_signup)
@api.doc(
responses={400: "error", 201: "error",}
)
def post(self):
json_data = request.get_json()
if not json_data:
return make_response(jsonify({"error": "no data"}), 400)
...
user = User(email=email,role=role,confirmed=confirmed,password=password,first_name=first_name, last_name=last_name, tel_number=tel_number_formatted,active=active)
try:
db.session.add(user)
db.session.commit()
except:
return make_response(
jsonify({"error": "cant't add user to data base"}), 400
)
return make_response(
jsonify(
{
"success": "user created, mail confirmation sent",
"User id": user.id,
"User email": user.email,
}
),
201,
)
...
# File app.core.models.py
from app import db
from app import bcrypt
from flask_bcrypt import generate_password_hash, check_password_hash
import datetime
class User(db.Model):
__tablename__ = "users"
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String, unique=True, nullable=False)
first_name = db.Column(db.String, nullable=False)
last_name = db.Column(db.String, nullable=False)
password_hash = db.Column(db.String, nullable=False)
registered_on = db.Column(db.DateTime, nullable=False)
active = db.Column(db.Boolean, nullable=False)
admin = db.Column(db.Boolean, nullable=False, default=False)
tel_number = db.Column(db.String, nullable=False)
confirmed = db.Column(db.Boolean, nullable=False, default=False)
confirmed_on = db.Column(db.DateTime, nullable=True)
role = db.Column(db.String, nullable=False)
password_reset_date = db.Column(db.DateTime, nullable=True)
last_login_on = db.Column(db.DateTime, nullable=True)
def __init__(
self,
email,
first_name,
last_name,
password,
confirmed,
active,
role,
tel_number,
admin=False,
confirmed_on=None,
):
self.email = email
self.first_name = first_name
self.last_name = last_name
self.password_hash = bcrypt.generate_password_hash(password).decode("utf-8")
self.registered_on = datetime.datetime.now()
self.admin = admin
self.tel_number = tel_number
self.role = role
self.confirmed = confirmed
self.confirmed_on = confirmed_on
self.active = active
def hash_password(self, password):
self.password_hash = bcrypt.generate_password_hash(password).decode("utf-8")
def check_password(self, password):
return check_password_hash(self.password_hash, password)
def __repr__(self):
return "<User {}>".format(self.email)
It works, but i had some errors with DB, which i can't repeat. So where my question is.
I use
db.session.add(user)
db.session.commit()
And for queries:
user = User.query.filter_by(email=email).first_or_404()
Must I use engine? Why? because it works without difining engine.. Must i use connections? I found the official documentation (https://docs.sqlalchemy.org/en/13/core/engines_connections.html) , but i don't really understand if it is necessary and what is the right way to use connections.
