| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283 |
- from fastapi import FastAPI, Form, Request
- from fastapi.encoders import jsonable_encoder
- from fastapi.middleware.cors import CORSMiddleware
- import dataset
- import pandas as pd
- from starlette.responses import FileResponse, StreamingResponse
- from fastapi.responses import HTMLResponse
- from fastapi.staticfiles import StaticFiles
- import io
- from starlette.responses import RedirectResponse
- app = FastAPI()
- app.add_middleware(
- CORSMiddleware,
- allow_origins=['*'],
- allow_credentials=True,
- allow_methods=["*"],
- allow_headers=["*"],
- )
- def get_db():
- db = dataset.connect('mysql://choozmo:pAssw0rd@139.162.121.30:33306/hhh?charset=utf8mb4')
- step_questions_table = db['step_questions']
- return step_questions_table
- @app.post("/step_questions/submit")
- async def submit(request: Request):
- form_data = await request.form()
- result = {
- 'sex': form_data.get('q1'),
- 'phone': form_data.get('q2'),
- 'email': form_data.get('q3'),
- 'building_case_name': form_data.get('q4'),
- 'building_case_type': form_data.get('q5'),
- 'decoration_style': ','.join(form_data.getlist('q6')),
- 'decoration_budget': ','.join(form_data.getlist('q7')),
- 'decoration_size': ','.join(form_data.getlist('q8')),
- }
- get_db().insert(result)
- response = RedirectResponse(url='/a1/index_complete.html')
- return response
- # return result
- app.mount("/a1", StaticFiles(directory="static"), name="static")
- @app.get("/a1")
- async def redirect():
- # url = app.url_path_for("/a1/index.html")
- response = RedirectResponse(url='/a1/index.html')
- return response
- #@app.get("/a1", response_class=HTMLResponse)
- #@app.get("/a1", response_class=HTMLResponse)
- #async def read_items():
- # return """
- # <html>
- # <head>
- # <title>Some HTML in here</title>
- # </head>
- # <body>
- # <h1>Look ma! HTML!</h1>
- # </body>
- # </html>
- # """
- @app.get("/step_questions")
- async def get_step_question(request: Request):
- db_data = list(get_db().find())
- df = pd.DataFrame(db_data, columns=db_data[0].keys())
- stream = io.StringIO()
- df.to_csv(stream, index = False)
- # response = StreamingResponse(data_frame.to_csv(index=False), media_type="text/csv")
- response = StreamingResponse(iter([stream.getvalue()]), media_type="text/csv")
- response.headers["Content-Disposition"] = "attachment; filename=export.csv"
- return response
|
