Mercurial > code > home > repos > video

comparison video.py @ 17:071943adf000

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
dnd a file or a url which we'll queue and fetch
author drewp@bigasterisk.com
date Sun, 16 Apr 2023 03:19:33 -0700
parents 53d99454f394
children 1b388ee5dd09
comparison
equal deleted inserted replaced
16:838eb0223bdb 17:071943adf000
1 import asyncio
2 import json
1 import logging 3 import logging
2 from pathlib import Path 4 from pathlib import Path
3 5
4 from prometheus_client import Gauge 6 from prometheus_client import Gauge
5 from starlette.applications import Starlette 7 from starlette.applications import Starlette
9 from starlette_exporter import PrometheusMiddleware, handle_metrics 11 from starlette_exporter import PrometheusMiddleware, handle_metrics
10 from sse_starlette.sse import EventSourceResponse 12 from sse_starlette.sse import EventSourceResponse
11 13
12 from video_file_store import VideoFileStore 14 from video_file_store import VideoFileStore
13 from video_ingest import VideoIngest 15 from video_ingest import VideoIngest
14 16 import dl_queue
15 logging.basicConfig(level=logging.DEBUG) 17 logging.basicConfig(level=logging.DEBUG)
16 log = logging.getLogger() 18 log = logging.getLogger()
19 logging.getLogger('sse_starlette').setLevel(logging.WARNING)
17 20
18 21
19 def root(req): 22 def root(req):
20 return HTMLResponse("api") 23 return HTMLResponse("api")
21 24
34 }) 37 })
35 38
36 39
37 async def ingestVideoUrl(req: Request) -> Response: 40 async def ingestVideoUrl(req: Request) -> Response:
38 url = await req.body() 41 url = await req.body()
39 svc.ingestUrl(url) 42 await svc.ingestUrl(url.decode('utf8'))
40 return Response(status_code=202) 43 return Response(status_code=202)
41 44
42 45
43 async def ingestVideoUpload(req: Request) -> Response: 46 async def ingestVideoUpload(req: Request) -> Response:
44 name = req.query_params['name'] 47 name = req.query_params['name']
45 await svc.addContent(name, req.body()) 48 await svc.addContent(name, req.body())
46 return Response(status_code=200) 49 return Response(status_code=200)
47 50
48 51
49 async def ingestQueue(req: Request) -> EventSourceResponse: 52 async def ingestQueue(req: Request) -> EventSourceResponse:
50 53 async def g():
51 def convertEvents(svcEvents): 54 async for ev in svc.events():
52 for ev in svcEvents: 55 yield json.dumps(ev)
53 yield dict(type='ev') 56 return EventSourceResponse(g())
54
55 return EventSourceResponse(convertEvents(svc.events()))
56 57
57 58
58 store = VideoFileStore(top=Path('/data')) 59 store = VideoFileStore(top=Path('/data'))
59 svc = VideoIngest(store) 60 svc = VideoIngest(store)
60
61 61
62 def main(): 62 def main():
63 63
64 app = Starlette( 64 app = Starlette(
65 debug=True, 65 debug=True,
76 ], 76 ],
77 ) 77 )
78 78
79 app.add_middleware(PrometheusMiddleware, app_name='video_api') 79 app.add_middleware(PrometheusMiddleware, app_name='video_api')
80 app.add_route("/metrics", handle_metrics) 80 app.add_route("/metrics", handle_metrics)
81
82 app.state.processTask = asyncio.create_task(dl_queue.process())
81 return app 83 return app
82 84
83 85
84 app = main() 86 app = main()