-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
70 lines (54 loc) · 1.86 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
"""
Author: Daniel W. Jongepier
Purpose: Store contact information in a Postgresql database.
"""
from fastapi import FastAPI, Security, Depends, HTTPException
from fastapi.security.api_key import APIKeyHeader
from fastapi_crudrouter import SQLAlchemyCRUDRouter
from starlette.status import HTTP_403_FORBIDDEN
from db.postgres import SessionLocal, engine
from models.contact import ContactModel, Base
from schema.contacts import Contact, CreateContact
from config.settings import Settings
# Instantiate the settings
settings = Settings()
# Print the used settings to console:
# print(f"Database URI: {settings.database_uri} \n"
# f"API-Key: {settings.api_key}")
# Set the access token en key name.
API_KEY = settings.api_key
API_KEY_NAME = "access_token"
# Define the APIKeyHeader to check for.
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
# Create the metadata and bind to database engine.
Base.metadata.create_all(bind=engine)
# Instantiate the application based on FastAPI.
app = FastAPI()
# Function to open and close the database connection on use.
def get_db():
session = SessionLocal()
try:
yield session
session.commit()
finally:
session.close()
# Function to check if the ApiKey provided matches the key expected.
def key_auth(api_key_header: str = Security(api_key_header)):
if api_key_header == API_KEY:
return api_key_header
else:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Could not validate credentials"
)
# Define the CRUDRouter.
router = SQLAlchemyCRUDRouter(
schema=Contact,
create_schema=CreateContact,
db_model=ContactModel,
db=get_db,
prefix='contact',
delete_all_route=False,
dependencies=[Depends(key_auth)]
)
# Update the already instantiated FastAPI instance with the CRUDRouter.
app.include_router(router)