Falcon Hug более классно
Итак, последние несколько дней я ломал себе голову, чтобы обнять работу более классным способом без особой удачи. Может ли кто-нибудь показать мне быстрый пример того, как заставить мой код работать с hug? Документация для этого довольно скудна, поэтому я и спрашиваю.
api
|_ db
| |_ __init__.py
| |_ manager.py
| |_ models.py
|
|_ middleware
| |_ __init__.py
| |_ context.py
|
|_ resources
| |_ __init__.py
| |_ scores.py
|
|_ schemas
| |_ __init__.py
| |_ scores_creation.json
|
|_ __init__.py
|_ app.py
|_ __main__.py
А вот актуальные актуальные файлы:
#db/manager.py
import sqlalchemy
from sqlalchemy import orm
from sqlalchemy.orm import scoping
# Needed by the setup method as we want to make sure
# all models are loaded before we call create_all(...)
from db import models
class DBManager(object):
def __init__(self, connection=None):
self.connection = connection
self.engine = sqlalchemy.create_engine(self.connection)
self.DBSession = scoping.scoped_session(
orm.sessionmaker(
bind=self.engine,
autocommit=True
)
)
@property
def session(self):
return self.DBSession()
def setup(self):
# Normally we would add whatever db setup code we needed here.
# This will for fine for the ORM
try:
models.SAModel.metadata.create_all(self.engine)
except Exception as e:
print('Could not initialize DB: {}'.format(e))
# db/models.py
import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base
SAModel = declarative_base()
class UserScores(SAModel):
__tablename__ = 'users'
id = sa.Column(sa.Integer, primary_key=True)
username = sa.Column(sa.String(128), unique=True)
company = sa.Column(sa.String(128))
score = sa.Column(sa.Integer)
def __init__(self, username, company, score):
self.username = username
self.company = company
self.score = score
@property
def as_dict(self):
return {
'username': self.username,
'company': self.company,
'score': self.score
}
def save(self, session):
with session.begin():
session.add(self)
@classmethod
def get_list(cls, session):
models = []
with session.begin():
query = session.query(cls)
models = query.all()
return models
# middleware/context.py
import uuid
import falcon
def set_context(req, resp):
if not req.context.get('request_id'):
req.context['request_id'] = str(uuid.uuid4())
resp.set_header('request-id', req.context['request_id'])
class ContextMiddleware(object):
def process_request(self, req, resp):
set_context(req, resp)
#resources/__init__.py
import falcon
import json
import jsonschema
class BaseResource(object):
def __init__(self, db_manager):
self.db = db_manager
def format_body(self, data):
return json.dumps(data)
def validate(schema):
def decorator(func):
def wrapper(self, req, resp, *args, **kwargs):
try:
raw_json = req.stream.read()
obj = json.loads(raw_json.decode('utf-8'))
except Exception:
raise falcon.HTTPBadRequest(
'Invalid data',
'Could not properly parse the provided data as JSON'
)
try:
jsonschema.validate(obj, schema)
except jsonschema.ValidationError as e:
raise falcon.HTTPBadRequest(
'Failed data validation',
e.message
)
return func(self, req, resp, *args, parsed=obj, **kwargs)
return wrapper
return decorator
# resources/scores.py
import falcon
from resources import BaseResource, validate
from schemas import load_schema
from sqlalchemy.exc import IntegrityError
from db import models
class ScoresResource(BaseResource):
def on_get(self, req, resp):
model_list = models.UserScores.get_list(self.db.session)
scores = [model.as_dict for model in model_list]
resp.status = falcon.HTTP_200
resp.body = self.format_body({
"scores": scores
})
@validate(load_schema('scores_creation'))
def on_post(self, req, resp, parsed):
model = models.UserScores(
username=parsed.get('username'),
company=parsed.get('company'),
score=parsed.get('score')
)
try:
model.save(self.db.session)
except IntegrityError:
raise falcon.HTTPBadRequest(
'Username exists',
'Could not create user due to username already existing'
)
resp.status = falcon.HTTP_201
resp.body = self.format_body({
'id': model.id
})
# schemas/__init__.py
import os
import json
def load_schema(name):
module_path = os.path.dirname(__file__)
path = os.path.join(module_path, '{}.json'.format(name))
with open(os.path.abspath(path), 'r') as fp:
data = fp.read()
return json.loads(data)
# schemas/scores_creation.json
{
"title": "Create Score",
"type": "object",
"properties": {
"username": {
"type": "string"
},
"company": {
"type": "string"
},
"score": {
"type": "integer",
"minimum": 0
}
},
"required": [
"username",
"company",
"score"
]
}
# api/app.py
import falcon
from db.manager import DBManager
from middleware.context import ContextMiddleware
from resources import scores
class MyService(falcon.API):
def __init__(self, cfg):
super(MyService, self).__init__(
middleware=[ContextMiddleware()]
)
self.cfg = cfg
# Build an object to manager our db connections.
mgr = DBManager(self.cfg.db.connection)
mgr.setup()
# Create our resources
scores_res = scores.ScoresResource(mgr)
# Build routes
self.add_route('/scores', scores_res)
def start(self):
""" A hook to when a Gunicorn worker calls run()."""
pass
def stop(self, signal):
""" A hook to when a Gunicorn worker starts shutting down. """
pass
# api/__main__.py
import aumbry
from docopt import docopt
from app import MyService
from gunicorn.app.base import BaseApplication
from gunicorn.workers.sync import SyncWorker
from config import AppConfig
class CustomWorker(SyncWorker):
def handle_quit(self, sig, frame):
self.app.application.stop(sig)
super(CustomWorker, self).handle_quit(sig, frame)
def run(self):
self.app.application.start()
super(CustomWorker, self).run()
class GunicornApp(BaseApplication):
""" Custom Gunicorn application
This allows for us to load gunicorn settings from an external source
"""
def __init__(self, app, options=None):
self.options = options or {}
self.application = app
super(GunicornApp, self).__init__()
def load_config(self):
for key, value in self.options.items():
self.cfg.set(key.lower(), value)
self.cfg.set('worker_class', '__main__.CustomWorker')
def load(self):
return self.application
def main():
docopt(__doc__)
cfg = aumbry.load(
aumbry.FILE,
AppConfig,
{
'CONFIG_FILE_PATH': 'config.yml'
}
)
api_app = MyService(cfg)
gunicorn_app = GunicornApp(api_app, cfg.gunicorn)
gunicorn_app.run()
main()
Как видите, я пропустил некоторые файлы конфигурации, так как не вижу, как они будут актуальны в этом примере. Моя проблема в том, что я не знаю точно, как загружать ресурсы на основе классов. Я попробовал использовать фрагменты, которые я выложил в таких разделах, как https://github.com/timothycrosley/hug/issues/464
Редактировать:
Я забыл упомянуть, что, конечно, я изменил проект, чтобы попытаться использовать @hug вместо стандартного сокола, но я не смог заставить его работать. То, что я разместил здесь, проект "без объятий", потому что, вероятно, было бы легче заставить его работать с объятиями из нормального проекта сокола