Skip to content

Completed Assessment #8

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ This is a single-file Python web server using the FastAPI web framework and the

This project is setup using `Pipenv`:
1. `pip install pipenv`
2. `pipenv install`
2. `pipenv install --skip-lock`
3. `pipenv run uvicorn index:app --reload`

### Submission
Expand Down
71 changes: 43 additions & 28 deletions index.py
Original file line number Diff line number Diff line change
@@ -1,58 +1,73 @@
from __future__ import print_function

import time

from typing import Any, Dict, Optional

from fastapi import FastAPI
from fastapi import FastAPI, Request
from fastapi.exceptions import HTTPException

from typing import Optional

from sqlmodel import Field, Session, SQLModel, create_engine, select


def get_current_timestamp():
current_timestamp: float = time.time() #e.g 1634600421.005548
return int(str(current_timestamp).split(".")[0])

class User(SQLModel, table=True):
# TODO change the default value to the timestamp that the user is created
id: Optional[int] = Field(default=None, primary_key=True)
# TODO change the default value to the timestamp that the user is created !DONE
# default_factory method should be used as a callable would be required to get the current_timestamp
id: Optional[int] = Field(
primary_key=True, default_factory=get_current_timestamp)
name: str
secret_name: str
age: Optional[str] = None

engine = create_engine("sqlite:///database.db")


DB_URL="sqlite:///database.db"
engine = create_engine(DB_URL)
SQLModel.metadata.create_all(engine)
# Create an instance of the API class
app = FastAPI()


@app.get("/")
async def root():
# TODO include the user's name if they are logged in
return {"message": "Hello {}".format("World")}
async def root(request: Request):
# TODO include the user's name if they are logged in !DONE
try:
return {"message": "Hello {}".format(request.user.name)}
except Exception:
return {"message": "Hello {}".format("World")}


@app.post("/user")
async def create_new_user(*, user: User):
with Session(engine) as session:
session.add(user)
session.commit()
# TODO return the User ID
return {"message": "User created"}

# deleting user provided id to use default id
user.id = None

# keep session alive for refresh after commit
with Session(engine, expire_on_commit=False) as session:
session.add(user)
session.commit()
# TODO return the User ID !DONE
return {"message": "User created", "user_id": user.id}


@app.get("/user/{id}")
async def get_user(id: int):
with Session(engine) as session:
# TODO return the user based on the ID (and an error if not)
statement = select(User).where(User.name == id)
user = session.exec(statement).first()
return {"user": user}
with Session(engine) as session:
# TODO return the user based on the ID (and an error if not) !DONE
# changed to query to look for the user.id instead of user.name
statement = select(User).where(User.id == id)
user = session.exec(statement).first()
if user:
return {"user": user}
return HTTPException(status_code=404, error="user_not_found")


@app.get("/api/webhooks")
async def handle_hook(*, event: Any):
id = event.payload.conversation.id
return {"message": "Hello World"}