Hobbit-core's API Documentation

hobbit cmd

hobbit - A flask project generator.

hobbit.bootstrap.new(*args, **kwargs) → Any

Create a new flask project, render from different template.

Examples:

hobbit --echo new -n blog -d /tmp/test -p 1024

It is recommended to use pipenv to create venv:

pipenv install -r requirements.txt && pipenv install --dev pytest pytest-cov pytest-env ipython flake8 ipdb

hobbit.bootstrap.gen(*args, **kwargs) → Any

Generator new feature. Auto gen models/{name}.py, schemas/{name}.py, views/{name}.py, services/{name.py}, tests/test_{name}.py etc.

hobbit_core

A flask extension that take care of base utils.

hobbit_core

Common utils for flask app.

class hobbit_core.HobbitManager(app=None, db=None, **kwargs)

Customizable utils management.

init_app(app, db, **kwargs)

app: The Flask application instance.

db

class hobbit_core.db.BaseModel(**kwargs)

Abstract base model class contains id、created_at and updated_at columns.

id: A surrogate integer 'primary key' column.

created_at: Auto save datetime.now() when row created.

updated_at: Auto save datetime.now() when row updated.

Support oracle id sequence, default name is {class_name}_id_seq, can changed by sequence_name and HOBBIT_UPPER_SEQUENCE_NAME config. Default value of app.config['HOBBIT_UPPER_SEQUENCE_NAME'] is False.

Examples:

from hobbit_core.db import Column, BaseModel

class User(BaseModel):
    username = Column(db.String(32), nullable=False, index=True)

print([i.name for i in User.__table__.columns])
# ['username', 'id', 'created_at', 'updated_at']

Can be blocked columns with exclude_columns:

class User(BaseModel):
    exclude_columns = ['created_at', 'updated_at']
    username = Column(db.String(32), nullable=False, index=True)

print([i.name for i in User.__table__.columns])
# ['username', 'id']

Can be changed primary_key's name using primary_key_name:

class User(BaseModel):
    primary_key_name = 'user_id'
    username = Column(db.String(32), nullable=False, index=True)

print([i.name for i in User.__table__.columns])
# ['username', 'user_id', 'created_at', 'updated_at']

Can be changed sequence's name using sequence_name (worked with oracle):

class User(BaseModel):
    sequence_name = 'changed'
    username = Column(db.String(32), nullable=False, index=True)

# print(User.__table__.columns['id'])
Column('id', ..., default=Sequence('changed_id_seq'))

__repr__() → str

You can set label property.

返回

<{classname}({pk}:{label!r})>

返回类型

str

class hobbit_core.db.SurrogatePK

A mixin that add id、created_at and updated_at columns to any declarative-mapped class.

id: A surrogate integer 'primary key' column.

created_at: Auto save datetime.now() when row created.

updated_at: Auto save datetime.now() when row updated.

It is not recommended. See hobbit_core.db.BaseModel.

__repr__() → str

You can set label property.

返回

<{classname}({pk}:{label!r})>

返回类型

str

class hobbit_core.db.EnumExt

Extension for serialize/deserialize sqlalchemy enum field.

Be sure type(key) is int and type(value) is str (label = (key, value)).

Examples:

class TaskState(EnumExt):
    # label = (key, value)
    CREATED = (0, '新建')
    PENDING = (1, '等待')
    STARTING = (2, '开始')
    RUNNING = (3, '运行中')
    FINISHED = (4, '已完成')
    FAILED = (5, '失败')

classmethod strict_dump(label: str, verbose: bool = False) → Union[int, str]

Get key or value by label.

Examples:

TaskState.strict_dump('CREATED')  # 0
TaskState.strict_dump('CREATED', verbose=True)  # '新建'

返回

Key or value, If label not exist, raise KeyError.

返回类型

int|str

classmethod dump(label: str, verbose: bool = False) → Dict[str, Any]

Dump one label to option.

Examples:

TaskState.dump('CREATED')  # {'key': 0, 'value': '新建'}

返回

Dict of label's key and value. If label not exist, raise KeyError.

返回类型

dict

classmethod load(val: Union[int, str]) → str

Get label by key or value. Return val when val is label.

Examples:

TaskState.load('FINISHED')  # 'FINISHED'
TaskState.load(4)  # 'FINISHED'
TaskState.load('新建')  # 'CREATED'

返回

Label.

返回类型

str|None

classmethod to_opts(verbose: bool = False) → List[Dict[str, Any]]

Enum to options.

Examples:

opts = TaskState.to_opts(verbose=True)
print(opts)

[{'key': 0, 'label': 'CREATED', 'value': u'新建'}, ...]

返回

List of dict which key is key, value, label.

返回类型

list

class hobbit_core.db.BaseModelMeta(name, bases, d)

class hobbit_core.db.BaseModel(**kwargs)

Abstract base model class contains id、created_at and updated_at columns.

id: A surrogate integer 'primary key' column.

created_at: Auto save datetime.now() when row created.

updated_at: Auto save datetime.now() when row updated.

Support oracle id sequence, default name is {class_name}_id_seq, can changed by sequence_name and HOBBIT_UPPER_SEQUENCE_NAME config. Default value of app.config['HOBBIT_UPPER_SEQUENCE_NAME'] is False.

Examples:

from hobbit_core.db import Column, BaseModel

class User(BaseModel):
    username = Column(db.String(32), nullable=False, index=True)

print([i.name for i in User.__table__.columns])
# ['username', 'id', 'created_at', 'updated_at']

Can be blocked columns with exclude_columns:

class User(BaseModel):
    exclude_columns = ['created_at', 'updated_at']
    username = Column(db.String(32), nullable=False, index=True)

print([i.name for i in User.__table__.columns])
# ['username', 'id']

Can be changed primary_key's name using primary_key_name:

class User(BaseModel):
    primary_key_name = 'user_id'
    username = Column(db.String(32), nullable=False, index=True)

print([i.name for i in User.__table__.columns])
# ['username', 'user_id', 'created_at', 'updated_at']

Can be changed sequence's name using sequence_name (worked with oracle):

class User(BaseModel):
    sequence_name = 'changed'
    username = Column(db.String(32), nullable=False, index=True)

# print(User.__table__.columns['id'])
Column('id', ..., default=Sequence('changed_id_seq'))

hobbit_core.db.reference_col(tablename: str, nullable: bool = False, pk_name: str = 'id', onupdate: str = None, ondelete: str = None, **kwargs) → sqlalchemy.sql.schema.Column

Column that adds primary key foreign key reference.

参数

• tablename (str) -- Model.__table_name__.

• nullable (bool) -- Default is False.

• pk_name (str) -- Primary column's name.

• onupdate (str) -- If Set, emit ON UPDATE <value> when issuing DDL for this constraint. Typical values include CASCADE, DELETE and RESTRICT.

• ondelete (str) -- If set, emit ON DELETE <value> when issuing DDL for this constraint. Typical values include CASCADE, DELETE and RESTRICT.

Others:

See sqlalchemy.Column.

Examples:

from sqlalchemy.orm import relationship

role_id = reference_col('role')
role = relationship('Role', backref='users', cascade='all, delete')

hobbit_core.db.transaction(session: sqlalchemy.orm.session.Session, nested: bool = False)

Auto transaction commit or rollback. This worked with session.autocommit=False (the default behavior of flask-sqlalchemy) or session.autocommit=True. See more: http://flask-sqlalchemy.pocoo.org/2.3/api/#sessions

Tips:

• Can't do session.commit() in func, otherwise raise sqlalchemy.exc.ResourceClosedError: This transaction is closed.

• Must use the same session in decorator and decorated function.

• We can use nested if keep top decorated by @transaction(session, nested=False) and all subs decorated by @transaction(session, nested=True).

Examples:

from hobbit_core.db import transaction

from app.exts import db


@bp.route('/users/', methods=['POST'])
@transaction(db.session)
def create(username, password):
    user = User(username=username, password=password)
    db.session.add(user)
    # db.session.commit() error occurred

We can nested use this decorator. Must set nested=True otherwise raise ResourceClosedError (session.autocommit=False) or raise InvalidRequestError (session.autocommit=True):

@transaction(db.session, nested=True)
def set_role(user, role):
    user.role = role
    # db.session.commit() error occurred


@bp.route('/users/', methods=['POST'])
@transaction(db.session)
def create(username, password):
    user = User(username=username, password=password)
    db.session.add(user)
    db.session.flush()
    set_role(user, 'admin')

pagination

hobbit_core.pagination.PageParams = {'order_by': <fields.DelimitedList(default=<marshmallow.missing>, attribute=None, validate=None, required=False, load_only=False, dump_only=False, missing=['-id'], allow_none=False, error_messages={'required': 'Missing data for required field.', 'null': 'Field may not be null.', 'validator_failed': 'Invalid value.', 'invalid': 'Not a valid list.'})>, 'page': <fields.Integer(default=<marshmallow.missing>, attribute=None, validate=<Range(min=1, max=2147483648, error=None)>, required=False, load_only=False, dump_only=False, missing=1, allow_none=False, error_messages={'required': 'Missing data for required field.', 'null': 'Field may not be null.', 'validator_failed': 'Invalid value.', 'invalid': 'Not a valid integer.', 'too_large': 'Number too large.'})>, 'page_size': <fields.Integer(default=<marshmallow.missing>, attribute=None, validate=<Range(min=5, max=100, error=None)>, required=False, load_only=False, dump_only=False, missing=10, allow_none=False, error_messages={'required': 'Missing data for required field.', 'null': 'Field may not be null.', 'validator_failed': 'Invalid value.', 'invalid': 'Not a valid integer.', 'too_large': 'Number too large.'})>}

Base params for list view func which contains page、page_size、 order_by params.

Example:

@use_kwargs(PageParams)
def list_users(page, page_size, order_by):
    pass

hobbit_core.pagination.pagination(obj: flask_sqlalchemy.model.DefaultMeta, page: int, page_size: int, order_by: Union[str, List[str], None] = 'id', query_exp=None) → importlib._bootstrap.PaginationType

A pagination for sqlalchemy query.

参数

• obj (db.Model) -- Model class like User.

• page (int) -- Page index.

• page_size (int) -- Row's count per page.

• order_by (str, list, None) -- Example: 'id'、['-id', 'column_name'].

• query_exp (flask_sqlalchemy.BaseQuery) -- Query like User.query.filter_by(id=1).

返回

Dict contains items、page、page_size and total fileds.

返回类型

dict

schemas

utils

class hobbit_core.utils.ParamsDict

Just available update func.

Example:

@use_kwargs(PageParams.update({...}))
def list_users(page, page_size, order_by):
    pass

update(other=None)

Update self by other Mapping and return self.

class hobbit_core.utils.dict2object

Dict to fake object that can use getattr.

Examples:

In [2]: obj = dict2object({'a': 2, 'c': 3})

In [3]: obj.a
Out[3]: 2

In [4]: obj.c
Out[4]: 3

hobbit_core.utils.secure_filename(filename: str) → str

Borrowed from werkzeug.utils.secure_filename.

Pass it a filename and it will return a secure version of it. This filename can then safely be stored on a regular file system and passed to os.path.join().

On windows systems the function also makes sure that the file is not named after one of the special device files.

>>> secure_filename(u'哈哈.zip')
'哈哈.zip'
>>> secure_filename('My cool movie.mov')
'My_cool_movie.mov'
>>> secure_filename('../../../etc/passwd')
'etc_passwd'
>>> secure_filename(u'i contain cool ümläuts.txt')
'i_contain_cool_umlauts.txt'

hobbit_core.utils.use_kwargs(argmap, schema_kwargs: Optional[Dict[KT, VT]] = None, **kwargs)

For fix Schema(partial=True) not work when used with @webargs.flaskparser.use_kwargs. More details see webargs.core.

参数

• argmap (marshmallow.Schema,dict,callable) -- Either a marshmallow.Schema, dict of argname -> marshmallow.fields.Field pairs, or a callable that returns a marshmallow.Schema instance.

• schema_kwargs (dict) -- kwargs for argmap.

返回

A dictionary of parsed arguments.

返回类型

dict

hobbit_core.utils.import_subs(locals_, modules_only: bool = False) → List[str]

Auto import submodules, used in __init__.py.

参数

• locals -- locals().

• modules_only -- Only collect modules to __all__.

Examples:

# app/models/__init__.py
from hobbit_core.utils import import_subs

__all__ = import_subs(locals())

Auto collect Model's subclass, Schema's subclass and instance. Others objects must defined in submodule.__all__.

response

hobbit_core.response.gen_response(code: int, message: str = '', detail: Optional[str] = None) → importlib._bootstrap.RespType

Func for generate response body.

参数

• code (string, int) -- Extension to interact with web pages. Default is http response status_code like 200、404.

• message (string) -- For popup windows.

• detail (object) -- For debug, detail server error msg.

返回

A dict contains all args.

返回类型

dict

class hobbit_core.response.Result(response=None, status=None, headers=None, mimetype='application/json', content_type=None, direct_passthrough=False)

Base json response.

status = 200

class hobbit_core.response.SuccessResult(message: str = '', code: Optional[int] = None, detail: Any = None, status: Optional[int] = None)

Success response. Default status is 200, you can cover it by status arg.

status = 200

class hobbit_core.response.FailedResult(message: str = '', code: Optional[int] = None, detail: Any = None)

Failed response. status always 400.

status = 400

class hobbit_core.response.UnauthorizedResult(message: str = '', code: Optional[int] = None, detail: Any = None)

status = 401

class hobbit_core.response.ForbiddenResult(message: str = '', code: Optional[int] = None, detail: Any = None)

status = 403

class hobbit_core.response.ValidationErrorResult(message: str = '', code: Optional[int] = None, detail: Any = None)

status = 422

class hobbit_core.response.ServerErrorResult(message: str = '', code: Optional[int] = None, detail: Any = None)

status = 500

err_handler

class hobbit_core.err_handler.ErrHandler

Base error handler that catch all exceptions. Be sure response is:

{
    "code": "404",  # error code, default is http status code, you can change it
    "message": "Not found",  # for alert in web page
    "detail": "id number field length must be 18",  # for debug
}

Examples:

app.register_error_handler(Exception, ErrHandler.handler)