medical_assist/apps/extensions/init_sqlalhemy.py

import datetime
import os

from flask import Flask, request
from flask_sqlalchemy import SQLAlchemy
from flask_sqlalchemy.query import Query as BaseQuery
from flask_marshmallow import Marshmallow
from marshmallow import fields
from marshmallow.validate import (
    URL, Email, Range, Length, Equal, Regexp,
    Predicate, NoneOf, OneOf, ContainsOnly
)
from flask_migrate import Migrate


class Query(BaseQuery):
    def soft_delete(self):
        return self.update({"delete_at": datetime.datetime.now()})

    def logic_all(self):
        return self.filter_by(delete_at=None).all()

    def all_json(self, schema: Marshmallow().Schema):
        return schema(many=True).dump(self.all())

    def layui_paginate(self):
        return self.paginate(page=request.args.get('page', type=int),
                             per_page=request.args.get('limit', type=int),
                             error_out=False)

    def layui_paginate_json(self, schema: Marshmallow().Schema):
        """
        返回dict
        """
        _res = self.paginate(
            page=request.args.get('page', type=int),
            per_page=request.args.get('limit', type=int),
            error_out=False
        )
        return schema(many=True).dump(_res.items), _res.total, _res.page, _res.per_page

    def layui_paginate_db_json(self):
        """
        db.query(A.name).layui_paginate_db_json()
        """
        _res = self.paginate(page=request.args.get('page', type=int),
                             per_page=request.args.get('limit', type=int),
                             error_out=False)
        return [dict(i) for i in _res.items], _res.total

db = SQLAlchemy(query_class=Query)
ma = Marshmallow()

migrate = Migrate()

def init_databases(app: Flask):
    db.init_app(app)
    # db.create_all(app=app)
    ma.init_app(app)

    if os.environ.get('WERKZEUG_RUN_MAIN') == 'true':
        with app.app_context():
            try:
                db.engine.connect()
                # 导入sql
                # with app.open_resource('schema.sql', mode='r') as f:
                #     db.cursor().executescript(f.read())
                # db.commit()
            except Exception as e:
                exit(f"数据库连接失败: {e}")