Flask-Injector example

main.py

import boto3
from bloop import BaseModel, Column, String, Engine
from flask import Flask
from flask_injector import FlaskInjector, singleton
from flask_restful import Api, Resource
from injector import inject, provider
import random
import string
from bloop_local import patched_local_bloop_engine

app = Flask(__name__)
api = Api(app)


def get_random_user_name(N=4):
    return "".join(random.choices(string.ascii_uppercase + string.digits, k=N))


class User(BaseModel):
    name = Column(String, hash_key=True)

    def to_json(self):
        return {"name": self.name}


class Configuration:
    def __init__(self):
        self.url = ("stuff",)
        self.username = "mkay"
        self.dynamodb_url = "http://127.0.0.1:8000"
        self.table_prefix = "what-{table_name}"
        self.region_name = 'eu-west-1'


class DbContext:
    @inject
    def __init__(self, configuration: Configuration):
        self.configuration = configuration
        self.engine = patched_local_bloop_engine(
            table_name_template=self.configuration.table_prefix,
            endpoint=self.configuration.dynamodb_url,
        )
        self.engine.bind(User)
        # This is just to generate some boilerplace data in dynamodb and to slow down instantiation a bit
        record = User()
        record.name = get_random_user_name()
        self.engine.save(record)


class ApiHome(Resource):
    @inject
    def __init__(self, configuration: Configuration, db_context: DbContext):
        self.db_context = db_context
        self.configuration = configuration

    def get(self):
        all_users = self.db_context.engine.scan(User).all()
        all_users_ret = []
        for usr in all_users:
            all_users_ret.append(usr.to_json())
        return {
            "url": self.configuration.url,
            "username": self.configuration.username,
            "all_users": all_users_ret,
        }


api.add_resource(ApiHome, "/")


def configure(binder):
    binder.bind(Configuration)
    binder.bind(DbContext, scope=singleton)



FlaskInjector(app=app, modules=[configure])

app.run()