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_devstatic(app, modules_path)[source]¶

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¶

shopyo.api.database.autoload_models(verbose=False)[source]¶

Auto imports models from modules/ in desired file. Used so that flask_migrate does not miss models when migrating

Return type:: None

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¶

shopyo.api.enhance.base_context()[source]¶

Used to define global template values

Returns:: copy of dictionary
Return type:: dict

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.delete_file(path)[source]¶

shopyo.api.file.get_folders(path)[source]¶

shopyo.api.file.last_part_of_path(path)[source]¶

shopyo.api.file.path_exists(path)[source]¶

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

shopyo.api.file.tryrmtree(path, verbose=False)[source]¶

Tries to removes an entire directory tree. Path must point to a directory. Outputs message to stdin or stderr

Parameters:: path (string) – directory path to be removed
Returns:: returns true upon successful return false otherwise
Return type:: bool

shopyo.api.file.unique_filename(fname)[source]¶

api.forms¶

shopyo.api.forms.flash_errors(form)[source]¶

Auto flash errors from WKHtml forms Reqwires base module or similar notification mechanism

Parameters:: form (WKHtml form)
Return type:: None

api.html¶

shopyo.api.html.notify(message, alert_type='primary')[source]¶

Used with flash: flash(notify(‘blabla’))

Parameters:

message (str) – message to be displayed
alert_type (str) – bootstrap class

Return type:

None

shopyo.api.html.notify_danger(message)[source]¶

shopyo.api.html.notify_info(message)[source]¶

shopyo.api.html.notify_success(message)[source]¶

shopyo.api.html.notify_warning(message)[source]¶

api.info¶

shopyo.api.info.printinfo()[source]¶

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: object

Mixin that adds convenience methods for CRUD (create, read, update, delete) operations.

classmethod bulk_create(items: List[Dict[str, Any]]) → List[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) → T[source]¶

Create a new record and save it in the database.

Returns:: The created record
Return type:: T

delete(commit: bool = True) → T | None[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]

save(commit: bool = True) → T[source]¶

Save the record.

Parameters:: commit (bool) – Whether to commit the changes
Returns:: The saved record
Return type:: T

update(commit: bool = True, **kwargs) → T | None[source]¶

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.PaginationMixin[source]¶

Bases: object

Mixin that adds pagination functionality.

classmethod paginate(page: int = 1, per_page: int = 20) → Dict[str, Any][source]¶

Paginate records.

Parameters:

page (int) – Page number
per_page (int) – Number of items per page

Returns:

Dictionary containing paginated items and metadata

Return type:

Dict[str, Any]

class shopyo.api.models.PkModel(**kwargs: Any)[source]¶

Bases: YoModel

Base model class that includes CRUD convenience methods, plus adds a ‘primary key’ column named ‘id’.

classmethod bulk_create(items: List[Dict[str, Any]]) → List[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) → T¶

Create a new record and save it in the database.

Returns:: The created record
Return type:: T

delete(commit: bool = True) → T | None¶

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: int | str) → T | None[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: int | str) → 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 overriding query_class.

Warning

The query interface is considered legacy in SQLAlchemy. Prefer using session.execute(select()) instead.

query_class¶: alias of Query

registry = <sqlalchemy.orm.decl_api.registry object>¶

save(commit: bool = True) → T¶

Save the record.

Parameters:: commit (bool) – Whether to commit the changes
Returns:: The saved record
Return type:: T

update(commit: bool = True, **kwargs) → T | None¶

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.SearchMixin[source]¶

Bases: object

Mixin that adds search functionality.

classmethod search(query: str, fields: List[str]) → Query[source]¶

Search records by query string in specified fields.

Parameters:

query (str) – Search query
fields (List[str]) – List of field names to search in

Returns:

SQLAlchemy query with search filters

Return type:

Query

class shopyo.api.models.SoftDeleteMixin[source]¶

Bases: object

Mixin that adds soft delete functionality.

classmethod get_active() → Query[source]¶: Get all non-deleted records.

is_deleted = Column(None, Boolean(), table=None, nullable=False, default=ScalarElementColumnDefault(False))¶

soft_delete(commit: bool = True) → SoftDeleteMixin | None[source]¶: Soft delete the record by setting is_deleted to True.

class shopyo.api.models.TimestampMixin[source]¶

Bases: object

Mixin 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: object

Mixin that adds validation functionality.

is_valid() → bool[source]¶: Check if the model instance is valid.

validate() → List[str][source]¶

Validate the model instance.

Returns:: List of validation errors, empty if valid
Return type:: List[str]

class shopyo.api.models.YoModel(**kwargs: Any)[source]¶

Bases: CRUDMixin, Model

Base model class that includes CRUD convenience methods.

classmethod bulk_create(items: List[Dict[str, Any]]) → List[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) → T¶

Create a new record and save it in the database.

Returns:: The created record
Return type:: T

delete(commit: bool = True) → T | None¶

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 overriding query_class.

Warning

The query interface is considered legacy in SQLAlchemy. Prefer using session.execute(select()) instead.

query_class¶: alias of Query

registry = <sqlalchemy.orm.decl_api.registry object>¶

save(commit: bool = True) → T¶

Save the record.

Parameters:: commit (bool) – Whether to commit the changes
Returns:: The saved record
Return type:: T

update(commit: bool = True, **kwargs) → T | None¶

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¶

class shopyo.api.module.ModuleHelp(dunderfile, dundername)[source]¶

Bases: object

context()[source]¶

get_self_static(filename)[source]¶

method(methodname)[source]¶

redirect_url(url, **kwargs)[source]¶

render(filename, **kwargs)[source]¶: renders file.html found in module/templates/module/file.html

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]¶

Parameters:

name – Policy name (convention: "perm.<PERMISSION_NAME>").
check – Callable (user, resource=None) -> bool.

grant(role_name: str, *permissions: 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 Permission enum members.

has_permission(user, permission: 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 User instance (or proxy).
permission – The Permission enum member to check.
resource – Optional resource object for resource-aware policies.

Returns:

True if the user has the permission, False otherwise.

require(permission: 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 Permission required 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

shopyo.api.security.is_safe_redirect_url(target)[source]¶

Corresponds to Djangos is_safe_url

Parameters:: target (String) – url
Return type:: bool

shopyo.api.security.validate_csrf_token(token)[source]¶

Validate the CSRF token using constant-time comparison.

Parameters:: token (str) – The CSRF token to validate.
Returns:: True if the token is valid, False otherwise.
Return type:: bool

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

shopyo.api.templates.yo_render_string(template_string, context_dict)[source]¶

Renders a template from a string.

Usage: yo_render_string(“Hello {{ name }}!”, {“name”: “World”})

shopyo.api.templates.yo_safe(html_string)[source]¶

Marks a string as safe for rendering (prevents auto-escaping).

Usage: yo_safe(“<b>bold</b>”)

api.validators¶

shopyo.api.validators.get_module_path_if_exists(name)[source]¶

shopyo.api.validators.is_alpha_num_underscore(name)[source]¶

returns whether the given name contains only alphanumeric or underscore

Parameters:: name (str) – to value to check for alphanumeric or underscore
Returns:: returns True if name is alphanumeric, False otherwise
Return type:: bool

shopyo.api.validators.is_empty_str(string)[source]¶

shopyo.api.validators.is_valid_slug(text)[source]¶

shopyo.api.validators.is_valid_url(url)[source]¶

shopyo.api.validators.verify_slug(form, field)[source]¶

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[T])[source]¶

Bases: object

A 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: 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: 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

handle_update(item_id: int | str) → tuple[source]¶

Handle PUT/PATCH request for updating an item.

Parameters:: item_id (Union[int, str]) – Item ID
Returns:: Flask response tuple
Return type:: tuple

validate_request(data: Dict[str, Any]) → List[str][source]¶

Validate request data. Override this method in subclasses.

Parameters:: data (Dict[str, Any]) – Request data to validate
Returns:: List of validation errors, empty if valid
Return type:: List[str]

shopyo.api.endpoint.api_endpoint(model_class: Type[T])[source]¶

Decorator to create RESTful API endpoints for a model.

Parameters:: model_class (Type[T]) – The SQLAlchemy model class to use
Returns:: Decorated function that returns APIEndpoint instance
Return type:: Callable