tessl/pypi-flask-appbuilder

Simple and rapid application development framework, built on top of Flask, with detailed security, auto CRUD generation, and comprehensive UI components.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

REST API System

Name: tessl/pypi-flask-appbuilder
Author: tessl

Complete REST API framework with automatic CRUD operations, OpenAPI documentation, schema generation, and JSON serialization. The API system provides a comprehensive solution for building REST APIs with minimal code while maintaining full customization capabilities.

Capabilities

ModelRestApi Class

Full-featured REST API class providing complete CRUD operations with automatic schema generation, validation, filtering, pagination, and OpenAPI documentation.

from flask_appbuilder.api import ModelRestApi
from flask_appbuilder.models.sqla.interface import SQLAInterface

class ModelRestApi(BaseModelApi):
    def __init__(self):
        """Initialize REST API with automatic endpoint generation."""
    
    def info(self, **kwargs):
        """
        Get API metadata and schema information.
        
        GET /_info
        
        Returns:
        Dict with model schema, relationships, permissions, filters
        """
    
    def get_list(self, **kwargs):
        """
        Get paginated list of model instances.
        
        GET /
        
        Query parameters:
        - q: Rison encoded query filters
        - page_size: Number of items per page (max: max_page_size)
        - page: Page number (0-based)
        - order_column: Column to order by
        - order_direction: asc or desc
        
        Returns:
        Dict with count, ids, result array
        """
    
    def get(self, pk, **kwargs):
        """
        Get single model instance by primary key.
        
        GET /<pk>
        
        Parameters:
        - pk: Primary key value
        
        Returns:
        Dict with model data or 404 error
        """
    
    def post(self):
        """
        Create new model instance.
        
        POST /
        
        Request body: JSON with model data
        
        Returns:
        Dict with created model data and 201 status
        """
    
    def put(self, pk):
        """
        Update existing model instance.
        
        PUT /<pk>
        
        Parameters:
        - pk: Primary key value
        Request body: JSON with updated model data
        
        Returns:
        Dict with updated model data or 404/422 error
        """
    
    def delete(self, pk):
        """
        Delete model instance by primary key.
        
        DELETE /<pk>
        
        Parameters:
        - pk: Primary key value
        
        Returns:
        Dict with success message or 404 error
        """
    
    # Lifecycle hooks
    def pre_add(self, item):
        """
        Hook called before adding new item.
        
        Parameters:
        - item: Model instance to be added
        """
    
    def post_add(self, item):
        """
        Hook called after successfully adding item.
        
        Parameters:
        - item: Added model instance
        """
    
    def pre_update(self, item):
        """
        Hook called before updating item.
        
        Parameters:
        - item: Model instance to be updated
        """
    
    def post_update(self, item):
        """
        Hook called after successfully updating item.
        
        Parameters:
        - item: Updated model instance
        """
    
    def pre_delete(self, item):
        """
        Hook called before deleting item.
        
        Parameters:
        - item: Model instance to be deleted
        """
    
    def post_delete(self, item):
        """
        Hook called after successfully deleting item.
        
        Parameters:
        - item: Deleted model instance
        """
    
    def pre_get(self, data):
        """
        Hook called before returning single item data.
        
        Parameters:
        - data: Serialized item data dict
        
        Returns:
        Modified data dict
        """
    
    def pre_get_list(self, data):
        """
        Hook called before returning list data.
        
        Parameters:
        - data: Dict with count, ids, result
        
        Returns:
        Modified data dict
        """

# API configuration properties
datamodel = None                    # SQLAlchemy interface (required)
list_title = "List"                # API list title
show_title = "Show"                # API show title  
add_title = "Add"                  # API add title
edit_title = "Edit"                # API edit title

# Column configuration
list_columns = []                   # Columns returned in list view
show_columns = []                   # Columns returned in show view
add_columns = []                    # Columns accepted for creation
edit_columns = []                   # Columns accepted for updates
list_exclude_columns = []           # Columns excluded from list
show_exclude_columns = []           # Columns excluded from show  
add_exclude_columns = []            # Columns excluded from add
edit_exclude_columns = []           # Columns excluded from edit

# SQL optimization
list_select_columns = []            # Specific SELECT columns for list
show_select_columns = []            # Specific SELECT columns for show
list_outer_default_load = []        # Outer join loading for list
show_outer_default_load = []        # Outer join loading for show

# Pagination and ordering
page_size = 20                      # Default page size
max_page_size = 100                # Maximum allowed page size
order_columns = []                  # Columns available for ordering

# Validation and schema
description_columns = {}            # Column descriptions
validators_columns = {}             # Marshmallow validators dict
add_query_rel_fields = {}          # Related field queries for add
edit_query_rel_fields = {}         # Related field queries for edit
order_rel_fields = {}              # Related field ordering

# Schema customization  
list_model_schema = None           # Custom list schema class
add_model_schema = None            # Custom add schema class
edit_model_schema = None           # Custom edit schema class
show_model_schema = None           # Custom show schema class
model2schemaconverter = None       # Custom schema converter class

# Usage example
class PersonApi(ModelRestApi):
    datamodel = SQLAInterface(Person)
    list_columns = ['id', 'name', 'email', 'created_on']
    show_columns = ['id', 'name', 'email', 'phone', 'created_on', 'updated_on']  
    add_columns = ['name', 'email', 'phone']
    edit_columns = ['name', 'email', 'phone']
    order_columns = ['name', 'email', 'created_on']
    description_columns = {
        'name': 'Person full name',
        'email': 'Contact email address'
    }

# Register API
appbuilder.add_api(PersonApi)

BaseApi Class

Foundation class for building custom API endpoints with security, routing, and response handling capabilities.

from flask_appbuilder.api import BaseApi, expose, safe

class BaseApi:
    def __init__(self):
        """Initialize API with base configuration."""
    
    def create_blueprint(self, appbuilder, endpoint=None, static_folder=None):
        """
        Create Flask blueprint for API.
        
        Parameters:
        - appbuilder: AppBuilder instance
        - endpoint: Custom endpoint name
        - static_folder: Static files folder
        
        Returns:
        Flask Blueprint instance
        """
    
    def add_api_spec(self, api_spec):
        """
        Add OpenAPI specification to API documentation.
        
        Parameters:
        - api_spec: APISpec instance
        """
    
    def add_apispec_components(self, api_spec):
        """
        Add custom OpenAPI components.
        
        Parameters:
        - api_spec: APISpec instance
        """
    
    def get_method_permission(self, method_name):
        """
        Get permission name for API method.
        
        Parameters:
        - method_name: API method name
        
        Returns:
        Permission name string
        """
    
    # Standard HTTP responses
    def response(self, code, **kwargs):
        """
        Generic JSON response with status code.
        
        Parameters:
        - code: HTTP status code
        - **kwargs: Response data
        
        Returns:
        Flask Response with JSON content
        """
    
    def response_400(self, message="Bad request"):
        """Bad Request (400) response."""
    
    def response_401(self):  
        """Unauthorized (401) response."""
    
    def response_403(self):
        """Forbidden (403) response."""
    
    def response_404(self):
        """Not Found (404) response."""
    
    def response_422(self, message="Unprocessable Entity"):
        """Unprocessable Entity (422) response."""
    
    def response_500(self, message="Internal server error"):
        """Internal Server Error (500) response."""

# API configuration properties
endpoint = ""                      # API endpoint name
version = "v1"                     # API version
route_base = ""                    # API route base (auto-generated)
resource_name = ""                 # Resource name for OpenAPI
base_permissions = []              # Base permissions list
class_permission_name = ""         # Class permission name override
method_permission_name = {}        # Method permission name overrides
allow_browser_login = False        # Allow Flask-Login session auth
csrf_exempt = []                   # CSRF exempt methods list
apispec_parameter_schemas = {}     # Custom parameter schemas
openapi_spec_component_schemas = {} # OpenAPI component schemas
responses = {}                     # Standard HTTP responses dict
exclude_route_methods = set()      # Methods to exclude from routes
include_route_methods = set()      # Methods to include in routes
openapi_spec_methods = {}          # OpenAPI spec method overrides
openapi_spec_tag = {}             # OpenAPI tag configuration
limits = []                        # Rate limiting configuration

# Usage example
class CustomApi(BaseApi):
    route_base = "/custom"
    
    @expose('/hello/')
    @safe
    def hello(self):
        return self.response(200, message="Hello World")
    
    @expose('/protected/')  
    @protect()
    def protected_endpoint(self):
        return self.response(200, user=g.user.username)

# Register custom API  
appbuilder.add_api(CustomApi)

API Decorators

Decorators for API endpoint exposure, parameter parsing, error handling, and response processing.

from flask_appbuilder.api import expose, rison, safe, merge_response_func

@expose("/", methods=["GET", "POST"])
def expose(url="/", methods=("GET",)):
    """
    Expose API method as endpoint.
    
    Parameters:
    - url: URL pattern for endpoint
    - methods: HTTP methods list
    
    Usage:
    @expose('/users/', methods=['GET', 'POST'])
    @protect()
    def users_endpoint(self):
        if request.method == 'GET':
            return self.get_users()
        return self.create_user()
    """

@rison(schema=UserSchema())
def rison(schema=None):
    """
    Parse URI Rison arguments using Marshmallow schema.
    
    Parameters:
    - schema: Marshmallow schema for validation
    
    Usage:
    from marshmallow import Schema, fields
    
    class FilterSchema(Schema):
        name = fields.String()
        active = fields.Boolean()
    
    @expose('/filtered/')
    @rison(FilterSchema())
    @protect()
    def filtered_data(self, **kwargs):
        filters = kwargs.get('rison', {})
        return self.response(200, filters=filters)
    """

@safe
def safe(f):
    """
    Catch uncaught exceptions and return JSON error response.
    
    Usage:
    @expose('/may-fail/')
    @safe
    def risky_endpoint(self):
        # This will catch any unhandled exceptions
        # and return proper JSON error response
        result = risky_operation()
        return self.response(200, result=result)
    """

@merge_response_func(custom_processor, "data")  
def merge_response_func(func, key):
    """
    Merge response function results into endpoint response.
    
    Parameters:
    - func: Function to call with response data
    - key: Key name for merged data
    
    Usage:
    def add_metadata(data):
        return {"timestamp": datetime.utcnow().isoformat()}
    
    @expose('/with-meta/')
    @merge_response_func(add_metadata, "meta")
    @protect()
    def endpoint_with_meta(self):
        return self.response(200, result="success")
        # Response includes merged metadata
    """

BaseModelApi Class

Foundation for model-based APIs providing data model integration, filtering, and search capabilities without full CRUD operations.

from flask_appbuilder.api import BaseModelApi

class BaseModelApi(BaseApi):
    """Base class for model APIs with search and filter capabilities."""

# Model integration properties
datamodel = None                   # SQLAlchemy interface (required)
search_columns = []                # Searchable columns
search_filters = {}                # Custom search filters
search_exclude_columns = []        # Columns excluded from search
label_columns = {}                 # Column label overrides
base_filters = []                  # Base query filters  
base_order = []                    # Default query ordering

# Usage example - Read-only API
class ReadOnlyPersonApi(BaseModelApi):
    datamodel = SQLAInterface(Person)
    search_columns = ['name', 'email']
    base_filters = [['active', FilterEqual, True]]
    
    @expose('/search/')
    @protect()
    def search_persons(self):
        # Custom search implementation using datamodel
        query = self.datamodel.get_query()
        # Apply search logic
        return self.response(200, results=query.all())

OpenAPI Integration

Automatic OpenAPI (Swagger) documentation generation for all API endpoints with customizable schemas and documentation.

# OpenAPI is automatically enabled for all ModelRestApi endpoints
# Access documentation at: /api/v1/openapi.json
# Swagger UI available at: /swaggerview/

# Custom OpenAPI schemas
from marshmallow import Schema, fields

class PersonResponseSchema(Schema):
    id = fields.Integer()
    name = fields.String(description="Person's full name")
    email = fields.Email(description="Contact email address")
    created_on = fields.DateTime()

class PersonApi(ModelRestApi):
    datamodel = SQLAInterface(Person)
    
    # Custom response schema
    show_model_schema = PersonResponseSchema
    
    # Custom OpenAPI tags
    openapi_spec_tag = {
        "name": "Persons",
        "description": "Person management endpoints"
    }
    
    # Custom parameter schemas
    apispec_parameter_schemas = {
        "person_id": {
            "in": "path",
            "schema": {"type": "integer", "minimum": 1},
            "description": "Person ID"
        }
    }

# Configuration options
FAB_API_SWAGGER_UI = True          # Enable Swagger UI
FAB_API_SWAGGER_TEMPLATE = "appbuilder/swagger/swagger.html"
FAB_API_MAX_PAGE_SIZE = 100        # Maximum API page size

API Error Handling

Comprehensive error handling with structured JSON responses and proper HTTP status codes.

# Standard error response format
{
    "message": "Error description",
    "severity": "error",  # or "warning", "info"
    "details": {}         # Additional error details
}

# Common API errors and responses:

# 400 Bad Request - Invalid input
{
    "message": "Invalid request data",
    "severity": "error", 
    "details": {"field": "validation error message"}
}

# 401 Unauthorized - Authentication required
{
    "message": "Authentication required",
    "severity": "error"
}

# 403 Forbidden - Insufficient permissions
{
    "message": "Access denied", 
    "severity": "error"
}

# 404 Not Found - Resource not found
{
    "message": "Record not found",
    "severity": "error"
}

# 422 Unprocessable Entity - Validation errors
{
    "message": "Validation failed",
    "severity": "error",
    "details": {
        "email": ["Invalid email address"],
        "name": ["Name is required"]
    }
}

# Custom error handling in APIs
class PersonApi(ModelRestApi):
    def pre_add(self, item):
        if not item.email:
            raise ValidationError("Email is required")
    
    def post_add(self, item):
        try:
            send_welcome_email(item.email)
        except Exception as e:
            # Log error but don't fail the request
            log.error(f"Failed to send welcome email: {e}")

Install with Tessl CLI