anandtripathi5

# Elastic configs
es_read_conf = {
    "es.nodes" : "localhost",
    "es.port" : "9200",
    "es.resource" : "twitter/tweet"
}

es_write_conf = {
    "es.nodes" : "localhost",
    "es.port" : "9200",

anandtripathi5

from locust import HttpUser, task, between
from random import randint


class AwesomeApplication(HttpUser):
    wait_time = between(1, 5)

    @task
    def hello(self):
        self.client.get("/hello")

anandtripathi5

from flask import Flask, request, jsonify
from flask_jwt_extended import create_access_token
from flask_jwt_extended import get_jwt_identity
from flask_jwt_extended import jwt_required
from flask_jwt_extended import JWTManager

app = Flask('__name__')
# Setup the Flask-JWT-Extended extension
app.config["JWT_SECRET_KEY"] = "super-secret"  # Change this!
jwt = JWTManager(app)

anandtripathi5

from locust import HttpUser, task
from random import randint

class AwesomeApplication(HttpUser):
    @task
    def hello(self):
        self.client.get("/hello")
        
    @task
    def world(self):

anandtripathi5

from flask import Flask

from extensions import socketio


def create_app():
    app = Flask(__name__)

    # other initializations and configurations
    ...

anandtripathi5

import celery as celery

from extensions import socketio


@celery.task(name="some_task", bind=True)
def some_task(self, room_name=None, namespace=None):
    # Do some task
    # Send a event to room
    socketio.emit(

anandtripathi5

from flask_socketio import emit, join_room
from extensions import db

from base import BaseSocket


class Foo(BaseSocket):

    def on_get_foo_details(self, **kwargs):
        try:

anandtripathi5

from extensions import socketio
from resources.foo import Foo


def socket_init():
    socketio.on_namespace(Foo('/foo'))

anandtripathi5

from flask_socketio import emit
from extensions import db

from base import BaseSocket


class Foo(BaseSocket):

    def on_get_foo_details(self, **kwargs):
        try:

anandtripathi5

from flask_socketio import Namespace, disconnect
from flask import current_app as app

from extensions import db


class BaseSocket(Namespace):
    
    def on_connect(self):
        pass