Skip to content

Instantly share code, notes, and snippets.

@seta-phucpham
Last active January 25, 2024 04:05
Show Gist options
  • Save seta-phucpham/261485257a74279b372ab16008fe0cb0 to your computer and use it in GitHub Desktop.
Save seta-phucpham/261485257a74279b372ab16008fe0cb0 to your computer and use it in GitHub Desktop.
POC import / export
from fastapi import FastAPI, Depends, UploadFile
from fastapi.responses import StreamingResponse
import xlsxwriter
import io
from openpyxl import load_workbook
from openpyxl.worksheet.worksheet import Worksheet
from typing import List, Dict
class ExcelService:
def __init__(self):
pass
def import_file(self, file: str | io.BytesIO) -> List[Dict[str, str]]:
wb = load_workbook(file, data_only=True, read_only=True)
ws: Worksheet = wb.active
records: List[Dict[str, str]] = []
headers = next(ws.iter_rows(max_row=1, values_only=True))
for row in ws.iter_rows(min_row=2,values_only=True):
records.append(dict(zip(headers, row)))
return records
def export_file(self, headers: List[str], records: Dict[str, str]) -> io.BytesIO:
output = io.BytesIO()
workbook = xlsxwriter.Workbook(output, {"in_memory": True})
worksheet = workbook.add_worksheet()
for h_index, header in enumerate(headers):
worksheet.write(0, h_index, header)
for r_index, record in enumerate(records):
for h_index, header in enumerate(headers):
worksheet.write(r_index + 1, h_index, record[header])
workbook.close()
output.seek(0)
return output
users = [
{
"id": "6982c4ec-2566-4b41-bb95-1e375585a3b8",
"full_name": "Trần Tú Trang",
"email": "trang.tran@seta-international.vn",
"badge_number": "6878",
"is_active": True,
"refresh_token": "",
"zuid": None,
"settings": None,
},
{
"id": "a0c43c3e-8d92-48ba-a8c9-2fc96ea18d4f",
"full_name": "Bùi Viết Quyền",
"email": "quyenbv6970@seta-international.vn",
"badge_number": "6970",
"is_active": True,
"refresh_token": "",
"zuid": None,
"settings": None,
},
{
"id": "6e80665b-a62f-4a2e-9c19-de71f4b44a87",
"full_name": "Hoàng Minh Thái",
"email": "thai.hoang@seta-international.vn",
"badge_number": "6962",
"is_active": True,
"refresh_token": "",
"zuid": None,
"settings": None,
},
{
"id": "e0b696a3-a29c-4fde-afce-b01fca915132",
"full_name": "Trần Văn Ngọ",
"email": "ngo.tran@seta-international.vn",
"badge_number": "6966",
"is_active": True,
"refresh_token": "",
"zuid": None,
"settings": None,
},
{
"id": "7a144506-9cf4-40b8-b668-9b4e56211460",
"full_name": "Nguyễn Đức An",
"email": "an.nguyen@seta-international.vn",
"badge_number": "6976",
"is_active": True,
"refresh_token": "",
"zuid": None,
"settings": None,
},
{
"id": "4bcf62e6-3cd1-4b09-86c9-842b42b2f349",
"full_name": "Trần Thanh Tùng",
"email": "tung.tran@seta-international.vn",
"badge_number": "8962",
"is_active": True,
"refresh_token": "",
"zuid": None,
"settings": None,
},
{
"id": "7e4d8284-db2e-49f7-8f1e-7a2fbf668b30",
"full_name": "Nguyễn Thị Thùy Linh",
"email": "linh.nguyen1@seta-international.vn",
"badge_number": "6973",
"is_active": True,
"refresh_token": "",
"zuid": None,
"settings": None,
},
]
def get_excel_service():
return ExcelService()
app = FastAPI()
@app.get("/")
def read_root(service: ExcelService = Depends(get_excel_service)):
headers = ["id", "full_name", "email"]
result = service.export_file(headers=headers, records=users)
headers= {
"Content-Disposition": "attachment; filename=users.xlsx",
}
return StreamingResponse(result, headers=headers)
@app.post("/")
async def create_upload_file(file: UploadFile, service: ExcelService = Depends(get_excel_service)):
file.file.seek(0)
users = service.import_file(file=file.file)
return {"users": users}
-i https://pypi.org/simple
aiofiles==23.2.1; python_version >= '3.7'
annotated-types==0.6.0; python_version >= '3.8'
anyio==4.2.0; python_version >= '3.8'
click==8.1.7; python_version >= '3.7'
et-xmlfile==1.1.0; python_version >= '3.6'
fastapi==0.109.0; python_version >= '3.8'
h11==0.14.0; python_version >= '3.7'
httptools==0.6.1
idna==3.6; python_version >= '3.5'
openpyxl==3.1.2; python_version >= '3.6'
pydantic==2.5.3; python_version >= '3.7'
pydantic-core==2.14.6; python_version >= '3.7'
python-dotenv==1.0.1
python-multipart==0.0.6; python_version >= '3.7'
pyyaml==6.0.1
sniffio==1.3.0; python_version >= '3.7'
starlette==0.35.1; python_version >= '3.8'
typing-extensions==4.9.0; python_version >= '3.8'
uvicorn[standard]==0.27.0; python_version >= '3.8'
uvloop==0.19.0
watchfiles==0.21.0
websockets==12.0
xlsxwriter==3.1.9; python_version >= '3.6'
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment