Security News
Fluent Assertions Faces Backlash After Abandoning Open Source Licensing
Fluent Assertions is facing backlash after dropping the Apache license for a commercial model, leaving users blindsided and questioning contributor rights.
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.
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.
CAUTION: Atila is base on WSGI but can be run only with Skitai App Engine because it crosses the border of WSGI for async router.
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.
Atila almost fully support async/await
manner.
Briefly Atila has 2 event loops for:
asyncore
as main loopasyncio
as executorI still recommend use sync
funtions mainly unless massive
I/O related tasks or have no choice.
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
Type hinting enables IDE suggestions.
from atila import Atila, Context
def __mount__ (context: Context, app: Atila):
@app.route ("/ping")
def ping (context: Context):
return 'pong'
In this document, type hinting is not used for code readability.
For exmaple, you make local dumb search engine named as dumbseek
.
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))]
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']
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)
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.
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.
We have to create 3 endpoints:
POST /api/documents
for indexing documentGET /api/documents
for searching documentGET /api/documents/<int:doc_id>
for geting documentAdd __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.
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']
See https://gitlab.com/skitai/atila/-/tree/master/examples/dumbseek
dumbseek
as local libraryskitaid.py
, we can serve as RESTful API onlineFAQs
Atila Framework
We found that atila demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Fluent Assertions is facing backlash after dropping the Apache license for a commercial model, leaving users blindsided and questioning contributor rights.
Research
Security News
Socket researchers uncover the risks of a malicious Python package targeting Discord developers.
Security News
The UK is proposing a bold ban on ransomware payments by public entities to disrupt cybercrime, protect critical services, and lead global cybersecurity efforts.