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]

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

[vieyahn2017]

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

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

[vieyahn2017]

fastapi sqlmodel

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