Skip to content

Commit

Permalink
fix: api security
Browse files Browse the repository at this point in the history
  • Loading branch information
kamajus committed Jan 11, 2025
1 parent a07e165 commit 4a5bb69
Show file tree
Hide file tree
Showing 4 changed files with 73 additions and 69 deletions.
127 changes: 62 additions & 65 deletions api/main.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import datetime

from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Security
from fastapi import Depends, FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.templating import Jinja2Templates
Expand All @@ -24,7 +24,7 @@
"http://localhost:3000",
]

app = FastAPI()
app = FastAPI(dependencies=[Depends(get_api_key)])
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
Expand All @@ -37,82 +37,79 @@


@app.get("/invoice/{order_id}")
async def invoice(order_id: str, _=Security(get_api_key)):
try:
# Query the Invoice table for the specific order_id
order = dict(execute("SELECT * FROM Invoice WHERE invoice_id = ?", (order_id,)))
async def invoice(order_id: str):
# Query the Invoice table for the specific order_id
order = execute("SELECT * FROM Invoice WHERE invoice_id = ?", (order_id,))

if order is None:
raise HTTPException(status_code=404, detail="Invoice not found")
if order is None:
raise HTTPException(status_code=404, detail="Invoice not found")

client = dict(
execute("SELECT * FROM Client WHERE client_id = ?", (order["client_id"],))
)
order = dict(order)

invoice_products = execute(
"SELECT * from InvoiceProduct WHERE invoice_id = ?",
(order["invoice_id"],),
fetchall=True,
)
client = dict(
execute("SELECT * FROM Client WHERE client_id = ?", (order["client_id"],))
)

products = []
invoice_products = execute(
"SELECT * from InvoiceProduct WHERE invoice_id = ?",
(order["invoice_id"],),
fetchall=True,
)

for product in list(invoice_products):
product = dict(product)
products = []

product_data = dict(
execute(
"SELECT * FROM Product WHERE product_id = ?",
(product["product_id"],),
)
)
for product in list(invoice_products):
product = dict(product)

products.append(
{
"name": product_data["name"],
"unit_price": product_data["unit_price"],
"quantity": product["quantity"],
"total_price": product["total_price"],
}
product_data = dict(
execute(
"SELECT * FROM Product WHERE product_id = ?",
(product["product_id"],),
)

tt = datetime.datetime.strptime(order["created_at"], "%Y-%m-%dT%H:%M:%S.%fZ")
order["created_at"] = tt.strftime("%Y-%m-%d %H:%M")

# Prepare template context data
context_data = {
"name": client["name"],
"phone": client["phone"],
"address": client["address"],
"reference": order["reference"],
"created_at": order["created_at"],
"products": products,
"total_price": order["total_price"],
}

# Read the HTML template
with open(TEMPLATE_HTML_PATH) as template:
template_html = template.read()

# Render the template
template_url = render_template(
template_html, context_data, TEMPLATE_CSS, OUTPUT_FILENAME
)

# Create the PDF response
response = FileResponse(
template_url,
media_type="application/pdf",
headers={"Content-Disposition": f"inline; filename={OUTPUT_FILENAME}"},
products.append(
{
"name": product_data["name"],
"unit_price": product_data["unit_price"],
"quantity": product["quantity"],
"total_price": product["total_price"],
}
)
return response

except Exception as e:
# Log the error and raise an HTTPException
print(f"Error fetching invoice: {e}")
raise HTTPException(status_code=500, detail="Internal Server Error")
tt = datetime.datetime.strptime(order["created_at"], "%Y-%m-%dT%H:%M:%S.%fZ")
order["created_at"] = tt.strftime("%Y-%m-%d %H:%M")

# Prepare template context data
context_data = {
"name": client["name"],
"phone": client["phone"],
"address": client["address"],
"reference": order["reference"],
"created_at": order["created_at"],
"products": products,
"total_price": order["total_price"],
}

# Read the HTML template
with open(TEMPLATE_HTML_PATH) as template:
template_html = template.read()

# Render the template
template_url = render_template(
template_html, context_data, TEMPLATE_CSS, OUTPUT_FILENAME
)

# Create the PDF response
response = FileResponse(
template_url,
media_type="application/pdf",
headers={"Content-Disposition": f"inline; filename={OUTPUT_FILENAME}"},
)

return response


@app.get("/")
def root(_=Security(get_api_key)):
def root():
return {"message": "Welcome back!"}
5 changes: 5 additions & 0 deletions api/utils/secure.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,11 @@ def get_api_key(
Raises:
HTTPException: If the API key is invalid or missing.
"""
print(api_key_query in API_KEYS)
print(api_key_query in API_KEYS)

print(API_KEYS)

if api_key_query in API_KEYS:
return api_key_query
if api_key_header in API_KEYS:
Expand Down
3 changes: 3 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,9 @@ lint = "black api/ && isort api/"
run = "uvicorn api.main:app --reload"
seed = { cmd = "python api/database/seed.py" }

[tool.pytest.ini_options]
log_level = "ERROR"

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
7 changes: 3 additions & 4 deletions tests/conftest.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,11 @@
from os import getenv
from dotenv import load_dotenv
load_dotenv()

from os import getenv
from fastapi.testclient import TestClient
from pytest import fixture
from api.main import app

load_dotenv()


@fixture
def client():
return TestClient(app)
Expand Down

0 comments on commit 4a5bb69

Please sign in to comment.