Shopyo api#
Shopyo has some api which eases your life
api.assets#
- shopyo.api.assets.get_static(boxormodule, filename)[source]#
Generates url for static file. In Shopyo 2.0, this is a thin wrapper around the standard Flask static handler.
In Development (DEBUG=True): The request is intercepted by register_shopyo_static. In Production (DEBUG=False): The file is served from the flattened /static/modules/ folder.
- Parameters
boxormodule (String) – box or module name e.g. box__default/auth or someothermodule
filename (String) – path of the file inside the box or module
- Return type
URL for static file
- shopyo.api.assets.register_shopyo_static(app, modules_path)[source]#
Registers a transparent interceptor for module static files in debug mode. This allows ‘url_for(‘static’, filename=’modules/…’)’ to work dynamically during development without running collectstatic.
- Parameters
app (Flask app) –
modules_path (String) – path to the modules directory
api.constants#
api.database#
api.email#
This file email.py contains functions for sending text and html rendered emails asynchronously
- shopyo.api.email.send_async_email(to, subject, template, from_email=None, **kwargs)[source]#
Sends email anachronously i.e the function is non blocking. Assume email template is valid i.e it can be rendered using flask’ render_template function and both .html and .txt email template files exits
- Parameters
to (String) – recipient of the email
subject (String) – subject of the email
template (String) – template file path to be used in email body
from_email (String, optional) – sender of the email. If not set MAIL_DEFAULT_SENDER is used from config.
api.enhance#
api.file#
- shopyo.api.file.absdiroffile(filepath)[source]#
Gives absolute directory of file, normally expects __file__ as param
- Parameters
filepath (str) – path of file
- Returns
Absolute dir path of file
- Return type
str
- shopyo.api.file.trycopy(source, dest, verbose=False)[source]#
Non-ecursive copy of folder
- Parameters
source (str) – source folder path
dest (str) – destination folder path
- Return type
None
- shopyo.api.file.trycopytree(source, dest, verbose=False)[source]#
Recursive copy of folder
- Parameters
source (str) – source folder path
dest (str) – destination folder path
- Return type
None
- shopyo.api.file.trymkdir(path, verbose=False)[source]#
Creates dir at destination
- Parameters
path (str) – path with folder already in
- Return type
None
- shopyo.api.file.trymkfile(path, content, verbose=False)[source]#
Creates file
- Parameters
path (str) – path to create file with filename included
content (str) – file content
- Return type
None
- shopyo.api.file.tryrmcache(dir_name, verbose=False)[source]#
removes all __pycache__ starting from directory dir_name all the way to leaf directory
- Parameters
dir_name (string) – path from where to start removing pycache
- shopyo.api.file.tryrmfile(path, verbose=False)[source]#
tries to remove file path and output message to stdin or stderr. Path must point to a file
- Parameters
path (string) – path of the file to remove
- Returns
returns true upon successful removal false otherwise
- Return type
bool
api.forms#
api.html#
api.models#
Enhanced DB-related helper utilities for Flask applications. Provides powerful mixins and base classes for rapid development.
- class shopyo.api.models.CRUDMixin[source]#
Bases:
objectMixin that adds convenience methods for CRUD (create, read, update, delete) operations.
- classmethod bulk_create(items: List[Dict[str, Any]]) List[shopyo.api.models.T][source]#
Create multiple records at once.
- Parameters
items (List[Dict[str, Any]]) – List of dictionaries containing record data
- Returns
List of created records
- Return type
List[T]
- classmethod create(**kwargs) shopyo.api.models.T[source]#
Create a new record and save it in the database.
- Returns
The created record
- Return type
T
- delete(commit: bool = True) Optional[shopyo.api.models.T][source]#
Remove the record from the database.
- Parameters
commit (bool) – Whether to commit the changes
- Returns
The deleted record if committed, None otherwise
- Return type
Optional[T]
- class shopyo.api.models.PaginationMixin[source]#
Bases:
objectMixin that adds pagination functionality.
- class shopyo.api.models.PkModel(**kwargs: Any)[source]#
Bases:
shopyo.api.models.YoModelBase model class that includes CRUD convenience methods, plus adds a ‘primary key’ column named ‘id’.
- classmethod bulk_create(items: List[Dict[str, Any]]) List[shopyo.api.models.T]#
Create multiple records at once.
- Parameters
items (List[Dict[str, Any]]) – List of dictionaries containing record data
- Returns
List of created records
- Return type
List[T]
- classmethod create(**kwargs) shopyo.api.models.T#
Create a new record and save it in the database.
- Returns
The created record
- Return type
T
- delete(commit: bool = True) Optional[shopyo.api.models.T]#
Remove the record from the database.
- Parameters
commit (bool) – Whether to commit the changes
- Returns
The deleted record if committed, None otherwise
- Return type
Optional[T]
- classmethod get_by_id(record_id: Union[int, str]) Optional[shopyo.api.models.T][source]#
Get record by ID.
- Parameters
record_id (Union[int, str]) – ID of record to get
- Returns
Object identified by record_id if any, None otherwise
- Return type
Optional[T]
- classmethod get_or_404(record_id: Union[int, str]) shopyo.api.models.T[source]#
Get record by ID or raise 404 error.
- Parameters
record_id (Union[int, str]) – ID of record to get
- Returns
Object identified by record_id
- Return type
T
- Raises
werkzeug.exceptions.NotFound – If record not found
- id = Column(None, Integer(), table=None, primary_key=True, nullable=False)#
- metadata = MetaData()#
- query: t.ClassVar[Query]#
A SQLAlchemy query for a model. Equivalent to
db.session.query(Model). Can be customized per-model by overridingquery_class.Warning
The query interface is considered legacy in SQLAlchemy. Prefer using
session.execute(select())instead.
- query_class#
alias of
flask_sqlalchemy.query.Query
- registry = <sqlalchemy.orm.decl_api.registry object>#
- save(commit: bool = True) shopyo.api.models.T#
Save the record.
- Parameters
commit (bool) – Whether to commit the changes
- Returns
The saved record
- Return type
T
- update(commit: bool = True, **kwargs) Optional[shopyo.api.models.T]#
Update specific fields of a record.
- Parameters
commit (bool) – Whether to commit the changes
**kwargs – Fields to update
- Returns
The updated record if committed, None otherwise
- Return type
Optional[T]
- class shopyo.api.models.SoftDeleteMixin[source]#
Bases:
objectMixin that adds soft delete functionality.
- is_deleted = Column(None, Boolean(), table=None, nullable=False, default=ScalarElementColumnDefault(False))#
- soft_delete(commit: bool = True) Optional[shopyo.api.models.SoftDeleteMixin][source]#
Soft delete the record by setting is_deleted to True.
- class shopyo.api.models.TimestampMixin[source]#
Bases:
objectMixin that adds created_at and updated_at timestamps.
- created_at = Column(None, DateTime(), table=None, nullable=False, default=CallableColumnDefault(<function TimestampMixin.<lambda>>))#
- updated_at = Column(None, DateTime(), table=None, nullable=False, onupdate=CallableColumnDefault(<function TimestampMixin.<lambda>>), default=CallableColumnDefault(<function TimestampMixin.<lambda>>))#
- class shopyo.api.models.ValidationMixin[source]#
Bases:
objectMixin that adds validation functionality.
- class shopyo.api.models.YoModel(**kwargs: Any)[source]#
Bases:
shopyo.api.models.CRUDMixin,sqlalchemy.orm.decl_api.ModelBase model class that includes CRUD convenience methods.
- classmethod bulk_create(items: List[Dict[str, Any]]) List[shopyo.api.models.T]#
Create multiple records at once.
- Parameters
items (List[Dict[str, Any]]) – List of dictionaries containing record data
- Returns
List of created records
- Return type
List[T]
- classmethod create(**kwargs) shopyo.api.models.T#
Create a new record and save it in the database.
- Returns
The created record
- Return type
T
- delete(commit: bool = True) Optional[shopyo.api.models.T]#
Remove the record from the database.
- Parameters
commit (bool) – Whether to commit the changes
- Returns
The deleted record if committed, None otherwise
- Return type
Optional[T]
- metadata = MetaData()#
- query: t.ClassVar[Query]#
A SQLAlchemy query for a model. Equivalent to
db.session.query(Model). Can be customized per-model by overridingquery_class.Warning
The query interface is considered legacy in SQLAlchemy. Prefer using
session.execute(select())instead.
- query_class#
alias of
flask_sqlalchemy.query.Query
- registry = <sqlalchemy.orm.decl_api.registry object>#
- save(commit: bool = True) shopyo.api.models.T#
Save the record.
- Parameters
commit (bool) – Whether to commit the changes
- Returns
The saved record
- Return type
T
- update(commit: bool = True, **kwargs) Optional[shopyo.api.models.T]#
Update specific fields of a record.
- Parameters
commit (bool) – Whether to commit the changes
**kwargs – Fields to update
- Returns
The updated record if committed, None otherwise
- Return type
Optional[T]
api.module#
- shopyo.api.module.iter_modules(project_root)[source]#
Yields (module_name, module_path) for every valid module in the project, transparently handling both standalone modules and
'box__'nested modules.- Yields
module_name (str) – Dot-separated python path (e.g., ‘modules.box__shop.cart’) module_path (str): Absolute file system path to the module directory.
api.perms#
Policy-based authorization system replacing is_admin checks with a
Permission + PolicyEngine approach.
Usage:
from shopyo.api.perms import PolicyEngine, Permission
engine = PolicyEngine()
# Grant permissions to roles
engine.grant("admin", Permission.ADMIN_PANEL_ACCESS, Permission.USER_MANAGE)
engine.grant("editor", Permission.CONTENT_MANAGE)
# Define custom policy checks
engine.define("perm.CONTENT_PUBLISH", lambda user, resource: user.id == resource.owner_id)
# Use as a decorator on views
@engine.require(Permission.ADMIN_PANEL_ACCESS)
def dashboard():
return "Admin dashboard"
# Programmatic check
if engine.has_permission(current_user, Permission.USER_MANAGE):
...
- class shopyo.api.perms.Permission(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]#
Granular permissions for the Shopyo authorization system.
- Members:
USER_MANAGE: Create, update, delete users. USER_READ: View user profiles and lists. ROLE_MANAGE: Create, update, delete roles. SETTINGS_MANAGE: Modify application settings. CONTENT_MANAGE: Create, update, delete content. CONTENT_PUBLISH: Publish or approve content. DASHBOARD_VIEW: Access the admin dashboard. ADMIN_PANEL_ACCESS: Enter the admin panel at all.
- ADMIN_PANEL_ACCESS = 8#
- CONTENT_MANAGE = 5#
- CONTENT_PUBLISH = 6#
- DASHBOARD_VIEW = 7#
- ROLE_MANAGE = 3#
- SETTINGS_MANAGE = 4#
- USER_MANAGE = 1#
- USER_READ = 2#
- class shopyo.api.perms.PolicyEngine[source]#
Central authorization engine combining role-based and policy-based access.
Usage:
engine = PolicyEngine() engine.grant("admin", Permission.ADMIN_PANEL_ACCESS) engine.define("perm.CONTENT_PUBLISH", lambda u, r: u.id == r.owner_id) # Decorator on views @engine.require(Permission.ADMIN_PANEL_ACCESS) def dashboard(): ... # Programmatic check engine.has_permission(current_user, Permission.USER_MANAGE)
- define(name: str, check: Callable[[...], bool])[source]#
Register a custom policy check.
- Parameters
name – Policy name (convention:
"perm.<PERMISSION_NAME>").check – Callable
(user, resource=None) -> bool.
- grant(role_name: str, *permissions: shopyo.api.perms.Permission)[source]#
Assign one or more permissions to a role.
- Parameters
role_name – Name of the role (e.g.
"admin","editor").permissions – One or more
Permissionenum members.
- has_permission(user, permission: shopyo.api.perms.Permission, resource: Any = None) bool[source]#
Check whether a user holds a given permission.
- Evaluation order:
Admin users bypass all checks (transitional).
Any role the user belongs to is checked for the permission.
If a custom policy
perm.<PERMISSION_NAME>exists, it is invoked with(user, resource).
- Parameters
user – Flask-Login
Userinstance (or proxy).permission – The
Permissionenum member to check.resource – Optional resource object for resource-aware policies.
- Returns
Trueif the user has the permission,Falseotherwise.
- require(permission: shopyo.api.perms.Permission, resource: Any = None)[source]#
Decorator factory that protects routes with a permission check.
Returns 401 if the user is unauthenticated, 403 if the user lacks the required permission.
- Parameters
permission – The
Permissionrequired to access the route.resource – Optional resource passed to the policy check.
Example:
@blueprint.route("/admin") @engine.require(Permission.ADMIN_PANEL_ACCESS) def admin_index(): return "Admin panel"
- class shopyo.api.perms.Policy(name: str, check: Callable[[...], bool])[source]
A named policy with a callable check.
- name
Unique policy identifier (e.g.
"perm.CONTENT_PUBLISH").- Type
str
- check
Callable
(user, resource=None) -> bool.- Type
Callable[[…], bool]
api.security#
- shopyo.api.security.generate_csrf_token()[source]#
Generate a secure CSRF token.
Generates a secure CSRF token and stores it in the session if not present.
- Returns
The generated CSRF token.
- Return type
str
- shopyo.api.security.get_safe_redirect(url)[source]#
Returns url for root path if url not safe
- Parameters
url (String) – url
- Return type
url or root page
- shopyo.api.security.inject_csrf_token()[source]#
Inject CSRF token into Jinja2 template context.
- Returns
Dictionary containing the CSRF token for template context.
- Return type
dict
api.templates#
- shopyo.api.templates.yo_get_macro(template_name, macro_name)[source]#
Gets a macro from a template for use in Python code.
- Usage:
my_macro = yo_get_macro(“macros.html”, “my_macro”) html = my_macro(arg1, arg2)
- shopyo.api.templates.yo_render(template, context_dict)[source]#
Renders template.
Usage: yo_render(“index.html”, {“x”: 1, “y”: 2}) Same as render_template(“index.html”, x=1, y=2)
- Parameters
template (String) – template accessing
context_dict (Dict) – Template values
- Return type
html of template
api.validators#
api.response#
response.py
Utilities for standardized API responses in Flask.
- shopyo.api.response.error_response(message, status=400, errors=None, **kwargs)[source]#
Return a standard error JSON response.
- Parameters
message (str) – Error message.
status (int) – HTTP status code.
errors (dict or list, optional) – Additional error details.
**kwargs – Additional fields to include.
- Returns
JSON response.
- Return type
Flask Response
- shopyo.api.response.json_response(data=None, status=200, message=None, **kwargs)[source]#
Return a standard JSON response.
- Parameters
data (dict or list, optional) – The main data payload.
status (int) – HTTP status code.
message (str, optional) – Optional message.
**kwargs – Additional fields to include.
- Returns
JSON response.
- Return type
Flask Response
- shopyo.api.response.paginated_response(items, total, page, per_page, status=200, message=None, **kwargs)[source]#
Return a paginated JSON response.
- Parameters
items (list) – List of items for the current page.
total (int) – Total number of items.
page (int) – Current page number.
per_page (int) – Items per page.
status (int) – HTTP status code.
message (str, optional) – Optional message.
**kwargs – Additional fields to include.
- Returns
JSON response.
- Return type
Flask Response
api.endpoint#
endpoint.py
A powerful utility for building RESTful API endpoints in Flask applications. Provides decorators and base classes for rapid API development with common patterns.
- class shopyo.api.endpoint.APIEndpoint(model_class: Type[shopyo.api.endpoint.T])[source]#
Bases:
objectA utility class for building RESTful API endpoints with common patterns. Provides decorators and methods for handling CRUD operations, validation, and response formatting.
- format_response(data: Any) Dict[str, Any][source]#
Format response data. Override this method in subclasses.
- Parameters
data (Any) – Data to format
- Returns
Formatted response data
- Return type
Dict[str, Any]
- get_query_filters() Dict[str, Any][source]#
Get query filters from request args. Override this method in subclasses.
- Returns
Query filters
- Return type
Dict[str, Any]
- handle_create() tuple[source]#
Handle POST request for creating a new item.
- Returns
Flask response tuple
- Return type
tuple
- handle_delete(item_id: Union[int, str]) tuple[source]#
Handle DELETE request for removing an item.
- Parameters
item_id (Union[int, str]) – Item ID
- Returns
Flask response tuple
- Return type
tuple
- handle_get(item_id: Union[int, str]) tuple[source]#
Handle GET request for retrieving a single item.
- Parameters
item_id (Union[int, str]) – Item ID
- Returns
Flask response tuple
- Return type
tuple
- handle_list(page: int = 1, per_page: int = 20) tuple[source]#
Handle GET request for listing items with pagination.
- Parameters
page (int) – Page number
per_page (int) – Items per page
- Returns
Flask response tuple
- Return type
tuple