atila 0.26.11

Atila Framework

Atila

Atila is life-cycle hook based web framework which is run on Skitai WSGI App Engine.

# myservice/__init__.py
def __app__ ():
    from atila import Atila
    return Atila (__name__)

def __mount__ (context, app):
    @app.route ("/")
    def index (context):
        return "Hello, World"

# skitaid.py
import skitai
import myservice

skitai.mount ("/", myservice)
skitai.run ()

And run,

python3 skitaid.py

Now, http://localhost:5000/ is working.

Life-Cycle Hook Based Implementation

Users can use some special hook functions like __app__ (), __mount__ () and so on.

This hooks can be integrated your existing source codes without any side effects. You just add routed controllers and then it could work as API backend.

Important Notice

CAUTION: Atila is base on WSGI but can be run only with Skitai App Engine.

This means if you make your Atila app, you have no choice but Skitai as WSGI app server. And Atila's unique and unconventional style may become very hard work to port to other framework.

Async/Await Support

Atila almost fully support async/await manner.

Briefly Atila has 2 event loops for:

• handling requests with asyncore as main loop
• running async functions with asyncio as executor

I still recommend use sync funtions mainly unless massive I/O related tasks or have no choice.

Table of Content

• Atila
• Life-Cycle Hook Based Implementation
• Important Notice
• Async/Await Support
• Installation
• Quick Start
  • Dumb Seek
  • Unit Testing
  • Write Atila App Hooks
  • Add Launch Script
  • Add Analyzing API Endpoint
  • Add Indexing and Searching API Endpoints
  • Testing API
  • Final Source Codes
  • Conclusion
• Connecting Database
  • Option I: Database Management Not Required
  • Option II: Database Management Required: Django As ORM
    • Initialize Django Project
    • Auto Reloading For Development
    • Add Django App
    • Django Rest Framework
    • Working With Atila App
  • Conclusion
• App Life Cycle and Hooks
  • App Life Cycle and Hooks
    • Runtine Preference and Bootstrapping
  • Request Life Cycle and Hooks
• Context
  • Global Context
    • Proto Context
    • Thread Context
  • Request Context
    • Request Context
    • Cloned Context
  • Registering Context Scope Objects and Methods
• Routing
  • Routing
  • Request Parameters
  • Async Routing
• Request Parameters
  • URL / Body Parameter
  • Parameter Validation
• Processing Request
  • Request Object
    • Basic Members
    • Basic Methods
    • Route Options
  • Environment Variables
    • In The Template Engine
  • App & Request Gloabal
  • File Upload
  • Cookie
  • Session
    • Namespaced Session
  • Message Box
  • Route Name Based Call
  • Helpers
    • Conditional Prework
    • Checking Dependencies
• Making Response
  • Cache Control
  • HTTP Error
  • Primitive
    • String
    • API Response
    • Rendered Template
    • render_or_API
    • File
    • Static
    • Generator
    • Redirecting
    • RPC Response
    • Threaded Data Streaming
  • Building URL
• Websocket
  • Specifications
    • WS_STREAM
    • WS_SESSION
      • Opening/Closing Hooks
  • WebSocket Piped Process
• gRPC
  • Unary RPC
    • Async version
  • Async Streaming RPC
    • Response Streaming
    • Request Streaming
    • Bidirectional Streaming
• Access Control and Authentication
  • CORS (Cross Origin Resource Sharing) and Preflight
  • Custom Authentication
  • WWW-Authentication
    • Authentication On Specific Methods
    • Password Provider
    • Authentication On Entire App
  • Bearer Authentication
  • Test Passing
• Security and Token
  • Cross Site Request Forgery Token (CSRF Token)
  • JWT Token
  • One-Time Password
  • One-Time Token
• Event Bus
  • Request Life Cycle Events
• Interval Base App Maintenancing
• Multiple GPUs Allocation To Workers
• Using Task
  • Thread / Process / Subrocess
  • Tasks
• HTTP/2,3 Server Pushing
• Logging and Traceback
• Template Engine
  • Customizing Jinja2
  • Custom Error Templates
• Using Requests
• Working With Multiple Apps
  • Event Subscription
  • Data Exchanging
  • Accesing Other App Directly
• Apps Overriding
  • Conclusion
• Test Client
  • Integrating pytest and API Documentation
• Debugging
  • VS Code
• Deployment
  • Using Systemctl
  • Using AWS ECS (Elastic Container Service)
• Change Log

Installation

Requirements

Python 3.7+

Installation

Atila and other core base dependent libraries is developing on single milestone, install/upgrade all at once. Otherwise it is highly possible to meet some errors.

With pip

pip3 install -U atila skitai rs4

Optional required as you need,

pip3 install protobuf # for GRPC

<< Back To README

Quick Start

For exmaple, you make local dumb search engine named as dumbseek.

Dumb Seek

Your package structure is like this.

dumbseek/
    index/
        __init__.py
        indexer.py
        searcher.py
        db.py
    __init__.py
    analyzer.py

File dumbseek/__init__.py

__version__ = "1.0"

NAME = "Dumb Seek"

File dumbseek/analyzer.py

def analyze (query):
    return query.lower ().split ()

File dumbseek/index/__init__.py is empty.

File dumbseek/index/db.py

INVERTED_INDEX = {}
DOCUMENTS = {}

File dumbseek/index/indexer.py

from .. import analyzer
from . import db

def index (doc):
    doc_ids = list (db.DOCUMENTS.keys ())
    if not doc_ids:
        doc_id = 0
    else:
        doc_id = max (doc_ids) + 1

    db.DOCUMENTS [doc_id] = doc
    for token in analyzer.analyze (doc):
        if token not in db.INVERTED_INDEX:
            db.INVERTED_INDEX [token] = set ()
        db.INVERTED_INDEX [token].add (doc_id)
    return doc_id

File dumbseek/index/searcher.py

from .. import analyzer
from . import db

def search (query):
    results = None
    for token in analyzer.analyze (query):
        if token not in db.INVERTED_INDEX:
            return []
        doc_ids = db.INVERTED_INDEX.get (token, set ())
        if results is None:
            results = doc_ids
            continue
        results = results.intersection (doc_ids)
    return [db.DOCUMENTS [doc_id] for doc_id in sorted (list (results))]

Unit Testing

For more clarify, pytest example is:

import os
import sys; sys.path.insert (0, '../examples/dumbseek')
import dumbseek
from dumbseek import analyzer
from dumbseek.index import indexer, searcher, db

def test_analyze ():
    assert analyzer.analyze ('Detective Holmes') == ['detective', 'holmes']

def test_index ():
    assert indexer.index ('Detective Holmes') == 0
    assert db.DOCUMENTS [0] == 'Detective Holmes'
    assert db.INVERTED_INDEX ['detective'] == {0}
    assert indexer.index ('Detective Monk') == 1
    assert db.INVERTED_INDEX ['monk'] == {1}
    assert searcher.search ('detective holmes') == ['Detective Holmes']

Write Atila App Hooks

Now, you find dumbseek is useful than you think, you have plan to serve with RESTful API.

Add __app__ and __mount__ hooks into dumbseek/__init__.py.

__version__ = "1.0"

NAME = "Dumb Seek"

# atila hooks ------------------------
def __app__ ():
    from atila import Atila
    return Atila (__name__)

def __mount__ (context, app):
    @app.route ("/")
    def index (context):
        return context.API (app = NAME)

Add Launch Script

For testing, you have to create launch script.

# skitaid.py
import skitai
import dumbseek

if __name__ == '__main__':
    with skitai.preference () as pref:
        skitai.mount ('/', dumbseek, pref)
    skitai.run (port = 5000, name = 'dumbseek')

python3 skitaid.py --devel

Now, http://localhost:5000/ is working.

Add Analyzing API Endpoint

We have to create endpoint POST /api/tokens. Add __mount__ hook to dumbseek/analyzer.py.

def analyze (query):
    return query.lower ().split ()

# atila hooks ------------------------
def __mount__ (context, app):
    @app.route ("/tokens", methods = ["POST", "OPTIONS"])
    def tokens (context, query):
        return context.API (result = analyze (query))

And for mounting to app, add __setup__ hook to dumbseek/__init__.py.

def __setup__ (context, app):
    from . import analyzer
    app.mount ("/api", analyzer)

def __mount__ (context, app):
    ...

http://localhost:5000/api/tokens is working.

Add Indexing and Searching API Endpoints

We have to create 3 endpoints:

• POST /api/documents for indexing document
• GET /api/documents for searching document
• GET /api/documents/<int:doc_id> for geting document

Add __mount__ to dumbseek/index/__init__.py.

# atila hooks ------------------------
def __mount__ (context, app):
    from . import indexer
    from . import searcher

    @app.route ("/documents", methods = ["POST", "OPTIONS"])
    def index_document (context, document):
        doc_id = indexer.index (document)
        return context.API (
            "201 Created",
            url = context.urlfor (get_document, doc_id)
        )

    @app.route ("/documents", methods = ["GET"])
    def search_document (context, q):
        return context.API (
            result = searcher.search (q)
        )

    @app.route ("/documents/<int:doc_id>", methods = ["GET"])
    def get_document (context, doc_id):
        try:
            return context.API (
                document = db.DOCUMENTS [doc_id]
            )
        except KeyError:
            raise context.HttpError ("404 Not Found")

And mount to app, add __setup__ hook to dumbseek/__init__.py.

def __setup__ (context, app):
    from . import analyzer
    from . import index
    app.mount ("/api", analyzer)
    app.mount ("/api", index)

def __mount__ (context, app):
    ...

http://localhost:5000/api/documents is working.

Testing API

import pytest
from functools import partial
import skitai
from atila.pytest_hooks import *

@pytest.fixture
def launch ():
    return partial (skitai.test_client, port = 30371, silent = False)

def test_api (launch):
    with launch ('../examples/dumbseek/skitaid.py') as engine:
        r = engine.get ('/')
        assert r.json () ['app'] == 'Dumb Seek'

        r = engine.post ('/api/tokens', data = {'query': 'Detective Holmes'})
        assert r.json () ['result'] == ['detective', 'holmes']

        r = engine.post ('/api/documents', data = {'document': 'Detective Holmes'})
        assert r.status_code == 201
        assert r.json () ['url'] == '/api/documents/0'

        r = engine.get ('/api/documents', params = {'q': 'Detective Holmes'})
        assert r.json () ['result'] == ['Detective Holmes']

        r = engine.get ('/api/documents/0')
        assert r.json ()['document'] == 'Detective Holmes'

        r = engine.post ('/api/documents', data = {'document': 'Detective Monk'})

        r = engine.get ('/api/documents', params = {'q': 'detective'})
        assert r.json () ['result'] == ['Detective Holmes', 'Detective Monk']

Final Source Codes

See https://gitlab.com/skitai/atila/-/tree/master/examples/dumbseek

Conclusion

• We add REST API to our existing source codes without any side effects
• We can still use dumbseek as local library
• Just add skitaid.py, we can serve as RESTful API online

<< Back To README

Connecting Database

If your project is database driven project, I recommend use Django and Django Rest Framework.

Also you can serve with Skitai App Engine instead of gunicorn or uvicorn:

import skitai
import sys; sys.path.insert (0, 'mydjangoproject')

skitai.mount ("/", "mydjangoproject/config/wsgi:application")
skitai.run ()

But if database is not important part of your project, it is worth to consider Atila.

Option I: Database Management Not Required

if you need quick building toy app with legacy database, use common database libraries like psycopg2, PyMySQL, cx-Oracle and etc.

This example is SQLPhile which is my own small package.

from sqlphile import pg2

def __setup__ (context, app):
    app.dbpool = pg2.Pool (200, "skitai", "skitai", "12345678", "localhost", 5432)

def __umounted__ (context, app):
    app.dbpool.close ()

def __mount__ (context, app):
    @app.route ("/transactions", methods = ['GET'])
    def transactions (context):
        with app.dbpool.acquire () as db:
            txlist = (db.select ("foo")
                        .filter (detail = 'ReturnTx')
                        .order_by ("-created_at")
                        .limit (10)
            ).execute ().fetch ()
        return context.API (txlist = txlist)

Option II: Database Management Required: Django As ORM

I personally loves Django ORM and admin views.

It is fine to develope Django and Atila seperately, and mount both apps. But for your convenience, I made some modification Django's manage.py.

And at project root, both manage.py and sktiaid.py are created for developing both Django and Atila apps.

Initialize Django Project

At your project root,

pip3 install -U atila django
wget https://gitlab.com/skitai/atila/-/raw/master/atila/collabo/django/manage.py
chmod +x manage.py
./manage.py startproject

I creates some Django related scripts at [project_root]/backend/orm which is integrated with Atila app.

If you change backend directory into myproject, modify manage.py and skitaid.py.

@customized_management ('myproject')
def main():
    os.environ.setdefault ('DJANGO_SETTINGS_MODULE', 'backend.myproject.settings')

...
import myproject

with skitai.preference () as pref:
    skitai.mount ('/', myproject, pref)
...

Then edit [project_root]/backend/orm/settings.py mostly DATABASE part.

./manage.py migrate
./manage.py createsuperuser # root:1234

Also [project_root]/skitaid.py is automatically generated.

chmod +x skitaid.py
./skitaid.py

You can access http://localhost:5000/ and http://localhost:5000/admin.

But page design and layout is not applied.

./manage.py collectstatic
./skitaid.py

Auto Reloading For Development

For developing, django usally run ./manage.py runserver. but it doses not work.

./skitaid.py --devel

Your sources changed, and restart process automatically.

Add Django App

From now, you can follow normal Django development procedure.

To add Django models,

./manage.py startapp blog

vi backend/orm/blog/models.py
vi backend/orm/blog/admin.py

# [project_root]/backend/orm/blog/models.py
from django.db import models

class Post (models.Model):
    title = models.CharField (max_length=200, unique=True)
    updated_on = models.DateTimeField (auto_now= True)
    content = models.TextField ()

Add backend.orm.blog app to settings.INSTALLED_APPS.

INSTALLED_APPS = [
    ...
    'django.contrib.staticfiles',
    'backend.orm.blog'
]

./manage.py makemigrations
./manage.py migrate

# [project_root]/backend/orm/blog/admin.py
from django.contrib import admin
from .models import Post

@admin.register (Post)
class PostAdmin (admin.ModelAdmin):
    list_display = ['id', 'title', 'updated_on']
    search_fields = ['title']

Run ./skitaid.py --devel again.

Django Rest Framework

# serializers.py
from rest_framework import serializers
from .models import Post

class PostSerializer (serializers.ModelSerializer):
    class Meta:
        model = Post
        fields = '__all__'

# [project_root]/backend/orm/blog/urls.py
from rest_framework import routers
from rest_framework import viewsets
from django.urls import path, include
from .serializers import PostSerializer
from .models import Post

class PostViewSet (viewsets.ModelViewSet):
    http_method_names = ['get', 'head']
    serializer_class = PostSerializer
    queryset = Post.objects.all ()

router = routers.DefaultRouter ()
router.register('posts', PostViewSet)
urlpatterns = [
    path('', include (router.urls))
]

# [project_root]/backend/urls.py
from django.contrib import admin
from django.urls import include, path
from backend.orm.blog import urls as blog_urls

urlpatterns = [
    path ('admin/', admin.site.urls),
    path ('api/', include (blog_urls))
]

Finally, collect static files for API veiw sets.

./manage.py collectstatic --noinput

Then your API works on http://localhost:5000/api/posts

Working With Atila App

[project_root]/backend/__init__.py is Atila app initiator and mount Django app.

BASE_DIR = os.path.dirname (__file__)

def __config__ (pref):
    import skitai
    from rs4 import pathtool
    from .orm import wsgi
    from django.conf import settings

    settings.DEBUG = skitai.is_devel ()
    pref.config.SETTINGS = settings
    pref.securekey = settings.SECRET_KEY
    skitai.mount ("/", wsgi.application, name = 'orm')

    for its in ('STATIC', 'MEDIA'):
        url, root = getattr (settings, f"{its}_URL"), getattr (settings, f"{its}_ROOT")
        pathtool.mkdir (root)
        skitai.mount (url, root)
        skitai.log_off (url)

def __app__ ():
    import atila
    return atila.Atila (__name__)

In Atila script, use Django ORM freely:

def __mount__ (context, app):
    from .orm.blog.models import Post

    @app.route ("/api/recent-posts")
    def index (context):
        return context.API (result = list (Post.objects.all ().values ()))

Conclusion

• Atila pawn ORM and admin pages off on Django even REST APIs
• We just build Atila APIs which is hardly concerned with database

<< Back To README