ivdenv backend (python)

[vieyahn2017]

ivdenv backend

[vieyahn2017]

def init_db(args):
    model.metadata.create_all(model.get_engine())
    model.start()

    if db_session.query(User).count() == 0:
        db_session.add(User({'NAME': 'admin', 'PASSWORD': 'Admin123'}))

    if db_session.query(Environment).count() == 0:
        db_session.add(Environment({'ID': 1, 'HOSTIP': '5.3.0.1', 'FLOATIP': '5.3.0.2', 'IIP': '8.4.2.9', 'NGINXIP': '5.3.7.0'}))

    model.commit()
    model.clear()

[vieyahn2017]

http://127.0.0.1:8090/rest/env

返回

{
  "data": [
    {
      "ivd": "https://xx",
      "ivdPackageInstallTime": "2023-07-29 11:58:44",

      "floatUser": "admin",
      "floatPort": "3194"
    }
  ],
  "error": {
    "code": 0,
    "description": "0"
  }
}

[vieyahn2017]

pecan框架封装的rest

class RestRoot(object):
    user = system_ctl.UsersCtl()
    env = system_ctl.EnvironmentsCtl()

    @expose(generic=True, template='json')
    def index(self):
        return {'data': {}, 'error': {'code': 0, 'description': '0'}}

自己的controller

class EnvironmentsCtl(object):
    @expose(generic=True, template='json')
    def index(self, **kwargs):
        data = IBaseCommon(Environment).get(**kwargs)
        return data

    @expose('json')
    def count(self, **kwargs):
        data = IBaseCommon(Environment).count(**kwargs)
        return data

    @index.when(template='json', method='POST')
    def post(self, **kwargs):
        obj = Environment(data)
        return ibase_jsonify(obj)

    @expose()
    def _lookup(self, obj_id, *remainder):
        obj = db_session.query(User).filter(User.ID == obj_id).first()
        if obj:
            return EnvironmentCtl(obj), remainder
        else:
            abort(200, json_body=ibase_jsonify({}, code=111))

class EnvironmentCtl(object):
    def __init__(self, obj):
        self.obj = obj

    @expose(generic=True, template='json')
    def index(self, **kwargs):
        return ibase_jsonify(self.obj)

    @index.when(template='json', method='PUT')
    def put(self, **kwargs):
        pass

    @index.when(template='json', method='DELETE')
    def delete(self):
        db_session.delete(self.obj)
        return ibase_jsonify({})

[vieyahn2017]

pecan @expose('json') def post

这是该框架的东西

[vieyahn2017]

Base 是封装了sqlalchemy的操作

Base = declarative_base(metadata=model.metadata)

参考 https://github.com/TransformerOptimus/SuperAGI/blob/9b455f56b655dc5ae2273f3a9e71e4154829ed2f/superagi/models/base_model.py#L10

class DBBaseModel(Base):
    """
    DBBaseModel is an abstract base class for all SQLAlchemy ORM models ,
    providing common columns and functionality.

    Attributes:
        created_at: Datetime column to store the timestamp about when a row is created.
        updated_at: Datetime column to store the timestamp about when a row is updated.

    Methods:
        to_dict: Converts the current object to a dictionary.
        to_json: Converts the current object to a JSON string.
        from_json: Creates a new object of the class using the provided JSON data.
        __repr__: Returns a string representation of the current object.
    """
    __abstract__ = True
    # id  = Column(INTEGER,primary_key=True,autoincrement=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    def to_dict(self):
        """
        Converts the current SQLAlchemy ORM object to a dictionary representation.

        Returns:
            A dictionary mapping column names to their corresponding values.
        """
        return {column.name: getattr(self, column.name) for column in self.__table__.columns}

    def to_json(self):
        """
            Converts the current SQLAlchemy ORM object to a JSON string representation.

            Returns:
                A JSON string representing the object with column names as keys and their corresponding values.
        """
        return json.dumps(self.to_dict())

    @classmethod
    def from_json(cls, json_data):
        """
            Creates a new SQLAlchemy ORM object of the class using the provided JSON data.

            Args: json_data (str): A JSON string representing the object with column names as keys and their
            corresponding values.

            Returns:
                A new SQLAlchemy ORM object of the class.
        """
        return cls(**json.loads(json_data))

    def __repr__(self):
        """
            Returns a string representation of the current SQLAlchemy ORM object.

            Returns:
                A string with the format "<Class Name> (<dictionary representation of the object>)".
        """
        return f"{self.__class__.__name__} ({self.to_dict()})"

[vieyahn2017]

app\model\system.py

from sqlalchemy import Column, Integer, String, Boolean

class Environment(Base):
    __tablename__ = 'environment'

    id = Column(String, primary_key=True)
    type = Column(Integer, default=100)
    cspHostIp = Column(String, default='')
    # ...
    description = Column(String, default='')
    users = Column(String, default='')
    modules = Column(String, default='')
    status = Column(String, default='0')

    def __init__(self, data):
        from_dict(self, data)

    def __json__(self):
        data = to_dict(self)
        return data

    def update(self, data):
        from_dict(self, data)

[vieyahn2017]

class IBaseCommon 封装了查询的filter

class IBaseCommon(object):
    def __init__(self, db_table):
        self.db_table = db_table

    def parse_filter(self, cond):
        LOG.debug(cond)
        if cond == 'and' or cond == 'or':
            return cond

    def filter(self, query, condition):
        pass

    def sort(self, query, condition):
        pass

    def range(self, condition):
        (start, end) = condition[1:-1].split('-')
        return start, end

    def query(self, **kwargs):
        LOG.debug(repr(kwargs))
        filter_cond = kwargs.pop('filter', None)
        order_cond = kwargs.pop('sortby', None)
        range_cond = kwargs.pop('range', None)

        query = db_session.query(self.db_table)
        data = query.all()
        return data

    def get(self, **kwargs):
        data = self.query(**kwargs)
        filter_cond = kwargs.pop('filter', None)
        if len(data) == 0 and filter_cond:
            return {"error": {"code": 0, "description": "0"}}
        else:
            return model.ibase_jsonify(data)

    def count(self, **kwargs):
        return model.ibase_jsonify({'COUNT': len(self.query(**kwargs))})

get调query，query调filter,sort等

==

现在看，这个框架也还是比较笨重啊？

[vieyahn2017]

后面看看找一个flask/fastapi的demo代替 简单要求： 1 restful 2 sqlite3

[vieyahn2017]

搜到一个这个 其实还是redirect直接渲染页面的，不符合我的要求

https://github.com/FaztWeb/flask-sqlite3-crud

from flask import Flask, render_template, request, url_for, redirect
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///database/tasks.db'

db = SQLAlchemy(app)

class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    content = db.Column(db.String(200))
    done = db.Column(db.Boolean)

@app.route('/')
def home():
    tasks = Task.query.all()
    return render_template('index.html', tasks = tasks)

@app.route('/create-task', methods=['POST'])
def create():
    new_task = Task(content=request.form['content'], done= False)
    db.session.add(new_task)
    db.session.commit()
    return redirect(url_for('home'))

@app.route('/done/<id>')
def done(id):
    task = Task.query.filter_by(id=int(id)).first()
    task.done = not(task.done)
    db.session.commit()
    return redirect(url_for('home'))

@app.route('/delete/<id>')
def delete(id):
    Task.query.filter_by(id=int(id)).delete()
    db.session.commit()
    return redirect(url_for('home'))

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

[vieyahn2017]

直接还跑不通，没有db文件 这是我补的脚本 把生成的database.db放到instance目录

import os
import sqlite3

SCHEMA_SQL = """
    CREATE TABLE task (
        id INTEGER,
        content TEXT,
        done INTEGER,
        PRIMARY KEY (id)
    );

"""

# 不需要单独建库文件，这样建的也不对
def mkdb(filename="database.db"):
    new = os.path.join(os.getcwd(), filename)
    if os.path.exists(new):
        return

    file = open(filename, 'w')
    file.write("\n")
    file.close()

# mkdb()

def init_db(sqlite_path="database.db"):
    # sqlite_path = 'sqlite:///database.db'
    sqlite_conn = sqlite3.connect(sqlite_path)
    sqlite_conn.executescript(SCHEMA_SQL)
    sqlite_conn.commit()
    sqlite_conn.close()

init_db()

上面的app.py改成 app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///students.sqlite3'

端口好像还占用了

if name == 'main':

app.run(debug=True)

app.run(debug=True,host="0.0.0.0",port=8080)

[vieyahn2017]

https://github.com/FaztWeb 这个很牛逼，很多项目，全是各种语言框架的最小demo

[vieyahn2017]

基本上算是找到了一个mini的，很好用

https://github.com/vieyahn2017/FastAPI-SQLModel-crash-course

[vieyahn2017]

fastapi sqlmodel

SQLModel 实际上是在 Pydantic 和 SQLAlchemy 之间增加了一层兼容适配，经过精心设计以兼容两者