Lukas Batteau lbatteau

## publish.py
@app.get("/sse/publish")
async def get(channel: str = "default", redis: Redis = Depends(depends_redis)):
    await redis.publish(channel=channel, message="Hello world!")
    return ""

## stream.py
@app.get("/sse/stream")
async def stream():
    return EventSourceResponse(subscribe)

## subscribe.py
async def subscribe(channel: str, redis: Redis) -> AsyncGenerator:
    (subscription,) = await redis.subscribe(channel=Channel(channel, False))
    while await subscription.wait_message():
        yield {"event": "message", "data": await subscription.get()}

## sse-fastapi-redis.py
from aioredis import Channel, Redis
from fastapi import FastAPI
from fastapi.params import Depends
from fastapi_plugins import depends_redis, redis_plugin
from sse_starlette.sse import EventSourceResponse
from starlette.responses import HTMLResponse

html = """
<!DOCTYPE html>
<html>

## README.markdown

      
        
          
            
              
              3 files
            
          
          
            
              
              0 forks
            
          
          
            
              
              0 comments
            
          
          
            
              
              1 star
            
          
        
        
          
              
          
          
            
                lbatteau
                / README.markdown
            
            
              Created
              October 30, 2012 12:38
                — forked from gudbergur/README.markdown
            
              
                Bootstrap's Typeahead plugin extended (allowing for AJAX functionality) among other things
              
          
        
      
        
  
      
    This is a fork of Bootstrap Typeahead that adds minimal but powerful extensions.

For example, process typeahead list asynchronously and return objects

  # This example does an AJAX lookup and is in CoffeeScript
  $('.typeahead').typeahead(
    # source can be a function
    source: (typeahead, query) ->
 # this function receives the typeahead object and the query string
	@app.get("/sse/publish")
	async def get(channel: str = "default", redis: Redis = Depends(depends_redis)):
	await redis.publish(channel=channel, message="Hello world!")
	return ""
	@app.get("/sse/stream")
	async def stream():
	return EventSourceResponse(subscribe)
	from aioredis import Channel, Redis
	from fastapi import FastAPI
	from fastapi.params import Depends
	from fastapi_plugins import depends_redis, redis_plugin
	from sse_starlette.sse import EventSourceResponse
	from starlette.responses import HTMLResponse

	html = """
	<!DOCTYPE html>
	<html>