Hobbit-core's API Documentation¶
hobbit cmd¶
hobbit - A flask project generator.
- hobbit.bootstrap.new(*args: Any, **kwargs: Any) 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_core¶
A flask extension that take care of base utils.
hobbit_core¶
Common utils for flask app.
db¶
- class hobbit_core.db.BaseModel(**kwargs)[源代码]¶
Abstract base model class contains
id
、created_at
andupdated_at
columns.id: A surrogate biginteger '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 bysequence_name
andHOBBIT_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'))
- class hobbit_core.db.SurrogatePK[源代码]¶
A mixin that add
id
、created_at
andupdated_at
columns to any declarative-mapped class.id: A surrogate biginteger '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.
- class hobbit_core.db.EnumExt(value)[源代码]¶
Extension for serialize/deserialize sqlalchemy enum field.
Be sure
type(key)
isint
andtype(value)
isstr
(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) # '新建'
- 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
.- 返回类型
- 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.
- 返回类型
- 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) # '新建'
- 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
.- 返回类型
- class hobbit_core.db.BaseModel(**kwargs)[源代码]¶
Abstract base model class contains
id
、created_at
andupdated_at
columns.id: A surrogate biginteger '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 bysequence_name
andHOBBIT_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'))
- query: t.ClassVar[Query]¶
A SQLAlchemy query for a model. Equivalent to
db.session.query(Model)
. Can be customized per-model by overridingquery_class
.警告
The query interface is considered legacy in SQLAlchemy. Prefer using
session.execute(select())
instead.
- hobbit_core.db.reference_col(tablename: str, nullable: bool = False, pk_name: str = 'id', onupdate: Optional[str] = None, ondelete: Optional[str] = None, **kwargs: Any) 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: Session, nested: bool = False)[源代码]¶
SQLAlchemy 1.4 deprecates “autocommit mode. See more: https://docs.sqlalchemy.org/en/14/orm/session_transaction.html
2022-05-18 Updated: Use nested=None to prevent signal bug, See more: https://github.com/pallets-eco/flask-sqlalchemy/issues/645
- Tips:
Can't do
session.commit()
in func,
otherwise unknown beloved.
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
We can nested use this decorator. Must set
nested=True
otherwise raiseResourceClosedError
(session.autocommit=False) or raiseInvalidRequestError
(session.autocommit=True):@transaction(db.session, nested=True) def set_role(user, role): user.role = role # db.session.commit() error @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(dump_default=<marshmallow.missing>, attribute=None, validate=None, required=False, load_only=False, dump_only=False, load_default=['-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 delimited list.'})>, 'page': <fields.Integer(dump_default=<marshmallow.missing>, attribute=None, validate=<Range(min=1, max=2147483648, min_inclusive=True, max_inclusive=True, error=None)>, required=False, load_only=False, dump_only=False, load_default=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(dump_default=<marshmallow.missing>, attribute=None, validate=<Range(min=5, max=100, min_inclusive=True, max_inclusive=True, error=None)>, required=False, load_only=False, dump_only=False, load_default=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
schemas¶
- class hobbit_core.schemas.ORMSchema(*args, **kwargs)[源代码]¶
Base schema for ModelSchema. See webargs/issues/126.
Example:
from hobbit_core.schemas import ORMSchema class UserSchema(ORMSchema): class Meta: model = User load_only = ('password')
@use_kwargs(UserSchema())
use in combination withload_only
:@bp.route('/users/', methods=['POST']) @use_kwargs(UserSchema()) def create_user(username, password): pass
- opts = <flask_marshmallow.sqla.SQLAlchemyAutoSchemaOpts object>¶
- class hobbit_core.schemas.SchemaMixin[源代码]¶
Add
id
,created_at
,updated_at
fields to schema, defaultdump_only=True
.Example:
from marshmallow import Schema from hobbit_core.schemas import SchemaMixin class UserSchema(Schema, SchemaMixin): pass
- class hobbit_core.schemas.PagedSchema(*, only: types.StrSequenceOrSet | None = None, exclude: types.StrSequenceOrSet = (), many: bool = False, context: dict | None = None, load_only: types.StrSequenceOrSet = (), dump_only: types.StrSequenceOrSet = (), partial: bool | types.StrSequenceOrSet = False, unknown: str | None = None)[源代码]¶
Base schema for list api pagination.
Example:
from marshmallow import fields from hobbit_core.schemas import PagedSchema from . import models from .exts import ma class UserSchema(ma.ModelSchema): class Meta: model = models.User class PagedUserSchema(PagedSchema): items = fields.Nested('UserSchema', many=True) paged_user_schemas = PagedUserSchema()
- class hobbit_core.schemas.ModelSchema(*args, **kwargs)[源代码]¶
Base ModelSchema for
class Model(db.SurrogatePK)
.Auto generate load and dump func for EnumField.
Auto dump_only for
id
,created_at
,updated_at
fields.Auto set dateformat to
'%Y-%m-%d %H:%M:%S'
.Auto use verbose for dump EnumField. See
db.EnumExt
. You can define verbose inMeta
.
Example:
class UserSchema(ModelSchema): role = EnumField(RoleEnum) class Meta: model = User data = UserSchema().dump(user).data assert data['role'] == {'key': 1, 'label': 'admin', 'value': '管理员'}
- opts = <flask_marshmallow.sqla.SQLAlchemyAutoSchemaOpts object>¶
utils¶
- class hobbit_core.utils.ParamsDict[源代码]¶
Just available update func.
Example:
@use_kwargs(PageParams.update({...})) def list_users(page, page_size, order_by): pass
- 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] = None, **kwargs: Any)[源代码]¶
For fix
Schema(partial=True)
not work when used with@webargs.flaskparser.use_kwargs
. More detailssee webargs.core
.
- 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__.
- hobbit_core.utils.bulk_create_or_update_on_duplicate(db, model_cls, items, updated_at='updated_at', batch_size=500)[源代码]¶
Support MySQL and postgreSQL. https://dev.mysql.com/doc/refman/8.0/en/insert-on-duplicate.html
- 参数
db -- Instance of SQLAlchemy.
model_cls -- Model object.
items -- List of data,[ example: [{key: value}, {key: value}, ...].
updated_at -- Field which recording row update time.
batch_size -- Batch size is max rows per execute.
- 返回
A dictionary contains rowcount and items_count.
- 返回类型
response¶
- hobbit_core.response.gen_response(code: int, message: Optional[str] = None, detail: Optional[str] = None, data=None) RespType [源代码]¶
Func for generate response body.
- 参数
- 返回
A dict contains all args.
- 返回类型
- 2021-07-08 Updated:
Default type of code in response is force conversion to str, now support set HOBBIT_USE_CODE_ORIGIN_TYPE = True to return origin type.
- 2021-07-13 Updated:
Support set HOBBIT_RESPONSE_MESSAGE_MAPS to use self-defined response message. HOBBIT_RESPONSE_MESSAGE_MAPS must be dict.
- class hobbit_core.response.Result(response=None, status=None, headers=None, mimetype='application/json', content_type=None, direct_passthrough=False)[源代码]¶
Base json response.
- response: Union[Iterable[str], Iterable[bytes]]¶
The response body to send as the WSGI iterable. A list of strings or bytes represents a fixed-length response, any other iterable is a streaming response. Strings are encoded to bytes as UTF-8.
Do not set to a plain string or bytes, that will cause sending the response to be very inefficient as it will iterate one byte at a time.
- class hobbit_core.response.SuccessResult(message: str = '', code: Optional[int] = None, detail: Optional[Any] = None, status: Optional[int] = None, data=None)[源代码]¶
Success response. Default status is 200, you can cover it by status arg.
- response: Union[Iterable[str], Iterable[bytes]]¶
The response body to send as the WSGI iterable. A list of strings or bytes represents a fixed-length response, any other iterable is a streaming response. Strings are encoded to bytes as UTF-8.
Do not set to a plain string or bytes, that will cause sending the response to be very inefficient as it will iterate one byte at a time.
- headers: Headers¶
- class hobbit_core.response.FailedResult(message: str = '', code: Optional[int] = None, detail: Optional[Any] = None)[源代码]¶
Failed response. status always 400.
- response: Union[Iterable[str], Iterable[bytes]]¶
The response body to send as the WSGI iterable. A list of strings or bytes represents a fixed-length response, any other iterable is a streaming response. Strings are encoded to bytes as UTF-8.
Do not set to a plain string or bytes, that will cause sending the response to be very inefficient as it will iterate one byte at a time.
- headers: Headers¶
- class hobbit_core.response.UnauthorizedResult(message: str = '', code: Optional[int] = None, detail: Optional[Any] = None)[源代码]¶
- response: Union[Iterable[str], Iterable[bytes]]¶
The response body to send as the WSGI iterable. A list of strings or bytes represents a fixed-length response, any other iterable is a streaming response. Strings are encoded to bytes as UTF-8.
Do not set to a plain string or bytes, that will cause sending the response to be very inefficient as it will iterate one byte at a time.
- headers: Headers¶
- class hobbit_core.response.ForbiddenResult(message: str = '', code: Optional[int] = None, detail: Optional[Any] = None)[源代码]¶
- response: Union[Iterable[str], Iterable[bytes]]¶
The response body to send as the WSGI iterable. A list of strings or bytes represents a fixed-length response, any other iterable is a streaming response. Strings are encoded to bytes as UTF-8.
Do not set to a plain string or bytes, that will cause sending the response to be very inefficient as it will iterate one byte at a time.
- headers: Headers¶
- class hobbit_core.response.ValidationErrorResult(message: str = '', code: Optional[int] = None, detail: Optional[Any] = None)[源代码]¶
- response: Union[Iterable[str], Iterable[bytes]]¶
The response body to send as the WSGI iterable. A list of strings or bytes represents a fixed-length response, any other iterable is a streaming response. Strings are encoded to bytes as UTF-8.
Do not set to a plain string or bytes, that will cause sending the response to be very inefficient as it will iterate one byte at a time.
- headers: Headers¶
- class hobbit_core.response.ServerErrorResult(message: str = '', code: Optional[int] = None, detail: Optional[Any] = None)[源代码]¶
- response: Union[Iterable[str], Iterable[bytes]]¶
The response body to send as the WSGI iterable. A list of strings or bytes represents a fixed-length response, any other iterable is a streaming response. Strings are encoded to bytes as UTF-8.
Do not set to a plain string or bytes, that will cause sending the response to be very inefficient as it will iterate one byte at a time.
- headers: Headers¶
err_handler¶
- exception hobbit_core.err_handler.HobbitException[源代码]¶
Base class for all hobbitcore-related errors.
- 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)