🐍 🌐 Web Framework Integration - Python
🌐 Web Framework Integration - Python
"We don't bow to any king" - Web Framework Edition
TuskLang seamlessly integrates with all major Python web frameworks, providing powerful configuration management, FUJSEN execution, and database integration capabilities.
🚀 Flask Integration
Basic Flask Setup
from flask import Flask, request, jsonify, render_template
from tsk import TSK
import osapp = Flask(__name__)
Load TuskLang configuration
config = TSK.from_file('app.tsk')@app.route('/')
def index():
return render_template('index.html',
app_name=config.get('app.name'),
version=config.get('app.version'))
@app.route('/api/process', methods=['POST'])
def process_data():
data = request.json
# Execute FUJSEN function for data processing
result = config.execute_fujsen(
'api',
'process_data',
data['amount'],
data['recipient']
)
return jsonify(result)
@app.route('/api/users')
def get_users():
# Use database query from TSK configuration
users = config.get('database.users_query')
return jsonify(users)
if __name__ == '__main__':
app.run(
host=config.get('server.host', '0.0.0.0'),
port=config.get('server.port', 5000),
debug=config.get('server.debug', False)
)
Advanced Flask Integration
from flask import Flask, request, jsonify, g
from tsk import TSK
from tsk.adapters import SQLiteAdapter
import functoolsapp = Flask(__name__)
Initialize TuskLang with database
tsk = TSK()
db_adapter = SQLiteAdapter('app.db')
tsk.set_database_adapter(db_adapter)Load configuration
config = tsk.from_file('app.tsk')def require_auth(f):
@functools.wraps(f)
def decorated_function(args, *kwargs):
# Use TSK for authentication logic
auth_result = config.execute_fujsen(
'auth',
'validate_token',
request.headers.get('Authorization')
)
if not auth_result['valid']:
return jsonify({'error': 'Unauthorized'}), 401
g.user = auth_result['user']
return f(args, *kwargs)
return decorated_function
@app.route('/api/payment', methods=['POST'])
@require_auth
def process_payment():
data = request.json
# Execute payment processing with FUJSEN
payment_result = config.execute_fujsen(
'payment',
'process_payment',
data['amount'],
data['recipient'],
g.user['id']
)
return jsonify(payment_result)
@app.route('/api/analytics')
@require_auth
def get_analytics():
# Use TSK for analytics queries
analytics = config.execute_fujsen(
'analytics',
'get_user_analytics',
g.user['id']
)
return jsonify(analytics)
Flask Configuration Management
app.tsk
app_config = """
$app_name: "MyFlaskApp"
$version: "1.0.0"[server]
host: "0.0.0.0"
port: @env("PORT", 5000)
debug: @env("DEBUG", false)
workers: @env("WORKERS", 1)
[database]
url: @env("DATABASE_URL", "sqlite:///app.db")
pool_size: @env("DB_POOL_SIZE", 10)
max_overflow: @env("DB_MAX_OVERFLOW", 20)
[security]
secret_key: @env("SECRET_KEY")
jwt_secret: @env("JWT_SECRET")
bcrypt_rounds: @env("BCRYPT_ROUNDS", 12)
[api]
rate_limit: @env("RATE_LIMIT", 100)
cache_ttl: @env("CACHE_TTL", 300)
[auth]
token_expiry: @env("TOKEN_EXPIRY", 3600)
refresh_token_expiry: @env("REFRESH_TOKEN_EXPIRY", 86400)
process_payment_fujsen = '''
def process_payment(amount, recipient, user_id):
# Validate amount
if amount <= 0:
raise ValueError("Amount must be positive")
# Check user balance
balance = query("SELECT balance FROM users WHERE id = ?", user_id)[0][0]
if amount > balance:
raise ValueError("Insufficient balance")
# Process payment
transaction_id = f"tx_{int(time.time())}"
# Update database
execute("UPDATE users SET balance = balance - ? WHERE id = ?", amount, user_id)
execute("INSERT INTO transactions (id, user_id, amount, recipient, created_at) VALUES (?, ?, ?, ?, ?)",
transaction_id, user_id, amount, recipient, time.time())
return {
'success': True,
'transaction_id': transaction_id,
'amount': amount,
'recipient': recipient
}
'''
validate_token_fujsen = '''
def validate_token(token):
if not token:
return {'valid': False, 'error': 'No token provided'}
try:
# Decode JWT token
payload = jwt.decode(token, jwt_secret, algorithms=['HS256'])
user_id = payload['user_id']
# Get user from database
user = query("SELECT id, username, email FROM users WHERE id = ?", user_id)
if not user:
return {'valid': False, 'error': 'User not found'}
return {
'valid': True,
'user': {
'id': user[0][0],
'username': user[0][1],
'email': user[0][2]
}
}
except jwt.ExpiredSignatureError:
return {'valid': False, 'error': 'Token expired'}
except jwt.InvalidTokenError:
return {'valid': False, 'error': 'Invalid token'}
'''
"""
🚀 Django Integration
Django Settings Integration
settings.py
from tsk import TSK
import osLoad TuskLang configuration
tsk_config = TSK.from_file('django.tsk')Django settings from TSK
SECRET_KEY = tsk_config.get('django.secret_key')
DEBUG = tsk_config.get('django.debug', False)
ALLOWED_HOSTS = tsk_config.get('django.allowed_hosts', ['*'])Database configuration
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': tsk_config.get('database.name'),
'USER': tsk_config.get('database.user'),
'PASSWORD': tsk_config.get('database.password'),
'HOST': tsk_config.get('database.host'),
'PORT': tsk_config.get('database.port'),
'OPTIONS': {
'sslmode': tsk_config.get('database.sslmode', 'require'),
},
}
}Email configuration
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = tsk_config.get('email.host')
EMAIL_PORT = tsk_config.get('email.port')
EMAIL_HOST_USER = tsk_config.get('email.user')
EMAIL_HOST_PASSWORD = tsk_config.get('email.password')
EMAIL_USE_TLS = tsk_config.get('email.use_tls', True)Cache configuration
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.redis.RedisCache',
'LOCATION': tsk_config.get('cache.redis_url'),
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.DefaultClient',
}
}
}Static files
STATIC_URL = tsk_config.get('static.url', '/static/')
STATIC_ROOT = tsk_config.get('static.root', 'staticfiles')
MEDIA_URL = tsk_config.get('media.url', '/media/')
MEDIA_ROOT = tsk_config.get('media.root', 'media')
Django Views with TuskLang
views.py
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods
from tsk import TSK
import jsonLoad TuskLang configuration
config = TSK.from_file('django.tsk')@csrf_exempt
@require_http_methods(["POST"])
def process_payment(request):
try:
data = json.loads(request.body)
# Execute payment processing with FUJSEN
result = config.execute_fujsen(
'payment',
'process_payment',
data['amount'],
data['recipient'],
request.user.id if request.user.is_authenticated else None
)
return JsonResponse(result)
except Exception as e:
return JsonResponse({'error': str(e)}, status=400)
@require_http_methods(["GET"])
def get_analytics(request):
# Use TSK for analytics
analytics = config.execute_fujsen(
'analytics',
'get_user_analytics',
request.user.id
)
return JsonResponse(analytics)
@require_http_methods(["GET"])
def get_config(request):
# Return configuration data
return JsonResponse({
'app_name': config.get('app.name'),
'version': config.get('app.version'),
'features': config.get('app.features', [])
})
Django Management Commands
management/commands/tsk_shell.py
from django.core.management.base import BaseCommand
from tsk import TSK
import codeclass Command(BaseCommand):
help = 'Start TuskLang shell with Django context'
def handle(self, args, *options):
# Load TuskLang configuration
config = TSK.from_file('django.tsk')
# Create shell with TuskLang context
shell_vars = {
'config': config,
'tsk': TSK(),
'execute_fujsen': config.execute_fujsen,
'get': config.get,
}
self.stdout.write(
self.style.SUCCESS('TuskLang shell started. Available variables:')
)
for var in shell_vars.keys():
self.stdout.write(f' - {var}')
# Start interactive shell
code.interact(local=shell_vars)
🚀 FastAPI Integration
Basic FastAPI Setup
from fastapi import FastAPI, HTTPException, Depends, Header
from fastapi.security import HTTPBearer
from pydantic import BaseModel
from tsk import TSK
from typing import Optional
import uvicornapp = FastAPI(title="TuskLang API", version="1.0.0")
Load TuskLang configuration
config = TSK.from_file('api.tsk')Security
security = HTTPBearer()Pydantic models
class PaymentRequest(BaseModel):
amount: float
recipient: str
description: Optional[str] = Noneclass UserResponse(BaseModel):
id: int
username: str
email: str
balance: float
Dependency for authentication
async def get_current_user(authorization: str = Depends(security)):
try:
# Use TSK for token validation
auth_result = config.execute_fujsen(
'auth',
'validate_token',
authorization.credentials
)
if not auth_result['valid']:
raise HTTPException(status_code=401, detail="Invalid token")
return auth_result['user']
except Exception as e:
raise HTTPException(status_code=401, detail=str(e))@app.post("/payment/process")
async def process_payment(
payment: PaymentRequest,
current_user: dict = Depends(get_current_user)
):
try:
result = config.execute_fujsen(
'payment',
'process_payment',
payment.amount,
payment.recipient,
current_user['id']
)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
raise HTTPException(status_code=500, detail="Internal server error")
@app.get("/users/me", response_model=UserResponse)
async def get_current_user_info(current_user: dict = Depends(get_current_user)):
# Get user details from TSK
user_info = config.execute_fujsen(
'user',
'get_user_info',
current_user['id']
)
return UserResponse(**user_info)
@app.get("/analytics")
async def get_analytics(current_user: dict = Depends(get_current_user)):
analytics = config.execute_fujsen(
'analytics',
'get_user_analytics',
current_user['id']
)
return analytics
if __name__ == "__main__":
uvicorn.run(
"main:app",
host=config.get('server.host', '0.0.0.0'),
port=config.get('server.port', 8000),
reload=config.get('server.debug', False)
)
Advanced FastAPI Integration
from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from tsk import TSK
import asyncio
import jsonapp = FastAPI()
Load configuration
config = TSK.from_file('api.tsk')CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=config.get('cors.allowed_origins', ['*']),
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)Background task processing
async def process_background_task(task_data: dict):
try:
result = config.execute_fujsen(
'background',
'process_task',
task_data
)
# Update task status
config.execute_fujsen(
'background',
'update_task_status',
task_data['task_id'],
'completed',
result
)
except Exception as e:
config.execute_fujsen(
'background',
'update_task_status',
task_data['task_id'],
'failed',
{'error': str(e)}
)@app.post("/tasks/create")
async def create_task(
task_data: dict,
background_tasks: BackgroundTasks,
current_user: dict = Depends(get_current_user)
):
# Create task
task_id = config.execute_fujsen(
'background',
'create_task',
task_data,
current_user['id']
)
# Add to background tasks
background_tasks.add_task(process_background_task, {
'task_id': task_id,
**task_data
})
return {"task_id": task_id, "status": "processing"}
@app.get("/tasks/{task_id}/status")
async def get_task_status(task_id: str):
status = config.execute_fujsen(
'background',
'get_task_status',
task_id
)
return status
Streaming response
@app.get("/data/stream")
async def stream_data():
async def generate():
for i in range(100):
data = config.execute_fujsen(
'streaming',
'generate_data',
i
)
yield f"data: {json.dumps(data)}\n\n"
await asyncio.sleep(0.1)
return StreamingResponse(generate(), media_type="text/plain")
🚀 Pyramid Integration
Basic Pyramid Setup
from pyramid.config import Configurator
from pyramid.response import Response
from pyramid.view import view_config
from tsk import TSK
import jsonLoad TuskLang configuration
config = TSK.from_file('pyramid.tsk')@view_config(route_name='home', renderer='json')
def home(request):
return {
'app_name': config.get('app.name'),
'version': config.get('app.version'),
'status': 'running'
}
@view_config(route_name='api_process', renderer='json', request_method='POST')
def process_data(request):
try:
data = request.json_body
result = config.execute_fujsen(
'api',
'process_data',
data['amount'],
data['recipient']
)
return result
except Exception as e:
request.response.status = 400
return {'error': str(e)}
@view_config(route_name='api_users', renderer='json')
def get_users(request):
users = config.get('database.users_query')
return {'users': users}
def main(global_config, **settings):
configurator = Configurator(settings=settings)
# Add routes
configurator.add_route('home', '/')
configurator.add_route('api_process', '/api/process')
configurator.add_route('api_users', '/api/users')
# Scan for views
configurator.scan()
return configurator.make_wsgi_app()
🚀 Bottle Integration
Basic Bottle Setup
from bottle import Bottle, request, response
from tsk import TSK
import jsonapp = Bottle()
Load TuskLang configuration
config = TSK.from_file('bottle.tsk')@app.route('/')
def index():
return {
'app_name': config.get('app.name'),
'version': config.get('app.version')
}
@app.route('/api/process', method='POST')
def process_data():
try:
data = request.json
result = config.execute_fujsen(
'api',
'process_data',
data['amount'],
data['recipient']
)
response.content_type = 'application/json'
return json.dumps(result)
except Exception as e:
response.status = 400
return json.dumps({'error': str(e)})
@app.route('/api/users')
def get_users():
users = config.get('database.users_query')
response.content_type = 'application/json'
return json.dumps({'users': users})
if __name__ == '__main__':
app.run(
host=config.get('server.host', 'localhost'),
port=config.get('server.port', 8080),
debug=config.get('server.debug', False)
)
🔧 Configuration Files
Flask Configuration (flask.tsk)
$app_name: "FlaskApp"
$version: "1.0.0"[server]
host: @env("HOST", "0.0.0.0")
port: @env("PORT", 5000)
debug: @env("DEBUG", false)
workers: @env("WORKERS", 1)
[database]
url: @env("DATABASE_URL", "sqlite:///app.db")
pool_size: @env("DB_POOL_SIZE", 10)
[security]
secret_key: @env("SECRET_KEY")
jwt_secret: @env("JWT_SECRET")
[api]
rate_limit: @env("RATE_LIMIT", 100)
cache_ttl: @env("CACHE_TTL", 300)
process_payment_fujsen = '''
def process_payment(amount, recipient, user_id):
if amount <= 0:
raise ValueError("Amount must be positive")
balance = query("SELECT balance FROM users WHERE id = ?", user_id)[0][0]
if amount > balance:
raise ValueError("Insufficient balance")
transaction_id = f"tx_{int(time.time())}"
execute("UPDATE users SET balance = balance - ? WHERE id = ?", amount, user_id)
execute("INSERT INTO transactions (id, user_id, amount, recipient) VALUES (?, ?, ?, ?)",
transaction_id, user_id, amount, recipient)
return {
'success': True,
'transaction_id': transaction_id,
'amount': amount,
'recipient': recipient
}
'''
Django Configuration (django.tsk)
$app_name: "DjangoApp"
$version: "1.0.0"[django]
secret_key: @env("SECRET_KEY")
debug: @env("DEBUG", false)
allowed_hosts: ["localhost", "127.0.0.1", @env("ALLOWED_HOST", "")]
time_zone: @env("TIME_ZONE", "UTC")
[database]
name: @env("DB_NAME", "django_app")
user: @env("DB_USER", "postgres")
password: @env("DB_PASSWORD", "")
host: @env("DB_HOST", "localhost")
port: @env("DB_PORT", 5432)
sslmode: @env("DB_SSLMODE", "require")
[email]
host: @env("EMAIL_HOST", "smtp.gmail.com")
port: @env("EMAIL_PORT", 587)
user: @env("EMAIL_USER", "")
password: @env("EMAIL_PASSWORD", "")
use_tls: @env("EMAIL_USE_TLS", true)
[cache]
redis_url: @env("REDIS_URL", "redis://localhost:6379/0")
[static]
url: "/static/"
root: "staticfiles"
[media]
url: "/media/"
root: "media"
process_payment_fujsen = '''
def process_payment(amount, recipient, user_id):
if amount <= 0:
raise ValueError("Amount must be positive")
balance = query("SELECT balance FROM users WHERE id = %s", user_id)[0][0]
if amount > balance:
raise ValueError("Insufficient balance")
transaction_id = f"tx_{int(time.time())}"
execute("UPDATE users SET balance = balance - %s WHERE id = %s", amount, user_id)
execute("INSERT INTO transactions (id, user_id, amount, recipient) VALUES (%s, %s, %s, %s)",
transaction_id, user_id, amount, recipient)
return {
'success': True,
'transaction_id': transaction_id,
'amount': amount,
'recipient': recipient
}
'''
FastAPI Configuration (api.tsk)
$app_name: "FastAPIApp"
$version: "1.0.0"[server]
host: @env("HOST", "0.0.0.0")
port: @env("PORT", 8000)
debug: @env("DEBUG", false)
workers: @env("WORKERS", 1)
[database]
url: @env("DATABASE_URL", "postgresql://user:pass@localhost/db")
pool_size: @env("DB_POOL_SIZE", 10)
max_overflow: @env("DB_MAX_OVERFLOW", 20)
[security]
secret_key: @env("SECRET_KEY")
jwt_secret: @env("JWT_SECRET")
algorithm: @env("JWT_ALGORITHM", "HS256")
[cors]
allowed_origins: ["http://localhost:3000", @env("FRONTEND_URL", "")]
allowed_credentials: true
[rate_limiting]
requests_per_minute: @env("RATE_LIMIT", 100)
burst_size: @env("BURST_SIZE", 10)
process_payment_fujsen = '''
def process_payment(amount, recipient, user_id):
if amount <= 0:
raise ValueError("Amount must be positive")
balance = query("SELECT balance FROM users WHERE id = %s", user_id)[0][0]
if amount > balance:
raise ValueError("Insufficient balance")
transaction_id = f"tx_{int(time.time())}"
execute("UPDATE users SET balance = balance - %s WHERE id = %s", amount, user_id)
execute("INSERT INTO transactions (id, user_id, amount, recipient) VALUES (%s, %s, %s, %s)",
transaction_id, user_id, amount, recipient)
return {
'success': True,
'transaction_id': transaction_id,
'amount': amount,
'recipient': recipient
}
'''
validate_token_fujsen = '''
def validate_token(token):
if not token:
return {'valid': False, 'error': 'No token provided'}
try:
payload = jwt.decode(token, jwt_secret, algorithms=[algorithm])
user_id = payload['user_id']
user = query("SELECT id, username, email FROM users WHERE id = %s", user_id)
if not user:
return {'valid': False, 'error': 'User not found'}
return {
'valid': True,
'user': {
'id': user[0][0],
'username': user[0][1],
'email': user[0][2]
}
}
except jwt.ExpiredSignatureError:
return {'valid': False, 'error': 'Token expired'}
except jwt.InvalidTokenError:
return {'valid': False, 'error': 'Invalid token'}
'''
🚀 Deployment Examples
Docker with Flask
FROM python:3.11-slimWORKDIR /app
Install TuskLang
RUN pip install tusklang flaskCopy application
COPY . .Copy TSK configuration
COPY flask.tsk /app/Expose port
EXPOSE 5000Run application
CMD ["python", "app.py"]
Docker with FastAPI
FROM python:3.11-slimWORKDIR /app
Install TuskLang and FastAPI
RUN pip install tusklang fastapi uvicornCopy application
COPY . .Copy TSK configuration
COPY api.tsk /app/Expose port
EXPOSE 8000Run application
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
Kubernetes Deployment
apiVersion: apps/v1
kind: Deployment
metadata:
name: tusk-flask-app
spec:
replicas: 3
selector:
matchLabels:
app: tusk-flask-app
template:
metadata:
labels:
app: tusk-flask-app
spec:
containers:
- name: app
image: tusk-flask-app:latest
ports:
- containerPort: 5000
env:
- name: APP_ENV
value: "production"
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: app-secrets
key: database-url
- name: SECRET_KEY
valueFrom:
secretKeyRef:
name: app-secrets
key: secret-key
volumeMounts:
- name: config
mountPath: /app/config
volumes:
- name: config
configMap:
name: app-config
---
apiVersion: v1
kind: Service
metadata:
name: tusk-flask-service
spec:
selector:
app: tusk-flask-app
ports:
- port: 80
targetPort: 5000
type: LoadBalancer
🔧 Testing Web Framework Integration
Flask Testing
import pytest
from flask import Flask
from tsk import TSK@pytest.fixture
def app():
app = Flask(__name__)
# Load test configuration
config = TSK.from_string("""
[test]
value: 42
string: "test"
test_fujsen = '''
def test_function(x):
return x * 2
'''
""")
app.config['TSK_CONFIG'] = config
return app
@pytest.fixture
def client(app):
return app.test_client()
def test_home_route(client):
response = client.get('/')
assert response.status_code == 200
assert b'FlaskApp' in response.data
def test_api_process(client):
response = client.post('/api/process',
json={'amount': 100, 'recipient': 'test@example.com'})
assert response.status_code == 200
data = response.get_json()
assert data['success'] == True
FastAPI Testing
import pytest
from fastapi.testclient import TestClient
from tsk import TSK@pytest.fixture
def client():
from main import app
return TestClient(app)
@pytest.fixture
def auth_headers():
# Create test token
config = TSK.from_file('api.tsk')
token = config.execute_fujsen('auth', 'create_token', {'user_id': 1})
return {'Authorization': f'Bearer {token}'}
def test_process_payment(client, auth_headers):
response = client.post(
'/payment/process',
json={'amount': 100, 'recipient': 'test@example.com'},
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert data['success'] == True
def test_get_user_info(client, auth_headers):
response = client.get('/users/me', headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert 'id' in data
assert 'username' in data
🎯 Best Practices
1. Configuration Management
- Use environment variables for sensitive data - Separate development and production configurations - Use TSK's hierarchical configuration for different environments2. Security
- Always validate input data - Use TSK's built-in SQL injection prevention - Implement proper authentication and authorization - Use HTTPS in production3. Performance
- Use TSK's caching capabilities - Implement proper database connection pooling - Use background tasks for heavy operations - Monitor and optimize FUJSEN execution4. Error Handling
- Implement comprehensive error handling - Use TSK's validation features - Log errors appropriately - Return meaningful error messages5. Testing
- Write unit tests for FUJSEN functions - Test web framework integration - Use TSK's testing utilities - Implement integration tests🚀 Next Steps
1. Choose your web framework (Flask, Django, FastAPI, etc.) 2. Set up TuskLang configuration for your framework 3. Implement FUJSEN functions for your business logic 4. Add database integration using TSK adapters 5. Deploy to production with proper configuration
---
"We don't bow to any king" - TuskLang integrates seamlessly with all major Python web frameworks, providing powerful configuration management, FUJSEN execution, and database integration capabilities. Choose your framework, configure with TSK, and build powerful web applications!