Basic functionality for the screenshot backend

This initial commit creates a FastAPI project with
some basic functionality for the screenshot
backend:

* Create users
* Basic authentication
* Create screenshot owned by the logged in user
* List all screenshots owned by the user
* Get a particular screenshot

.gitignore

@@ -0,0 +1,7 @@
+.idea
+venv/
+env/
+.env
+**/__pycache__
+.vscode
+Pipfile*

LICENSE

@@ -0,0 +1,19 @@
+MIT License
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,30 @@
+# Screenshot Capture Backend in FastAPI
+
+Do you find sometimes the need to share screenshots with friends? This project
+might be for you. You might also want to check the Chrome plugin in
+[this repo](https://github.com/sebasutp/screenshot-capture) to capture
+screenshots from Chrome and add them to the backend.
+
+## Want to test this project locally project?
+
+1. Fork/Clone
+
+1. Create and activate a virtual environment:
+
+    ```sh
+    $ python3 -m venv venv && source venv/bin/activate
+    ```
+
+1. Install the requirements:
+
+    ```sh
+    (venv)$ pip install -r requirements.txt
+    ```
+
+1. Run the app:
+
+    ```sh
+    (venv)$ python main.py
+    ```
+
+1. Test at [http://localhost:8081/docs](http://localhost:8081/docs)

app/api.py

@@ -0,0 +1,69 @@
+from fastapi import FastAPI, Body, Depends, HTTPException, status
+from typing import Annotated
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+
+from sqlmodel import Session, SQLModel, select
+from datetime import timedelta
+
+import app.model as model
+import app.auth.auth_handler as auth_handler
+import app.auth.crypto as crypto
+
+app_obj = FastAPI()
+
+@app_obj.on_event("startup")
+def on_startup():
+    model.create_db_and_tables()
+
+# route handlers
+
+@app_obj.post("/token")
+async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()], session: Session = Depends(model.get_db_session)):
+    user = model.UserLogin(email=form_data.username, password=form_data.password)
+    db_user = auth_handler.check_and_get_user(user, session)
+    if not db_user:
+        raise HTTPException(status_code=400, detail="Incorrect username or password")
+    token = auth_handler.create_access_token(db_user, timedelta(minutes=30))
+    return model.Token(access_token=token, token_type="bearer")
+
+
+@app_obj.get("/screenshots", tags=["screenshots"], response_model=list[model.Screenshot])
+async def get_screenshots(
+    *,
+    session: Session = Depends(model.get_db_session),
+    current_user_id: model.UserId = Depends(auth_handler.get_current_user_id)):
+    statement = select(model.Screenshot).where(model.Screenshot.owner_id == current_user_id.id)
+    return session.exec(statement)
+
+
+@app_obj.get("/screenshots/{id}", tags=["screenshots"], response_model=model.Screenshot)
+async def get_single_screenshot(*, session: Session = Depends(model.get_db_session), id: str):
+    q = select(model.Screenshot).where(model.Screenshot.external_id == id)
+    screenshot = session.exec(q).first()
+    if not screenshot:
+        raise HTTPException(status_code=404, detail="Screenshot not found")
+    return screenshot
+
+
+@app_obj.post("/screenshots", tags=["screenshots"], response_model=model.Screenshot)
+async def add_screenshots(
+    *,
+    session: Session = Depends(model.get_db_session),
+    current_user_id: model.UserId = Depends(auth_handler.get_current_user_id),
+    screenshot: model.ScreenshotCreate):
+    screenshot_db = model.Screenshot.model_validate(screenshot, update={"owner_id": current_user_id.id})
+    screenshot_db.external_id = crypto.generate_random_base64_string(32)
+    session.add(screenshot_db)
+    session.commit()
+    session.refresh(screenshot_db)
+    return screenshot_db
+
+@app_obj.post("/user/signup", tags=["user"])
+async def create_user(*, session: Session = Depends(model.get_db_session), user: model.UserCreate = Body(...)):
+    db_user = model.User.model_validate(user)
+    # Hash password before saving it
+    db_user.password = crypto.get_password_hash(db_user.password)
+    session.add(db_user)
+    session.commit()
+    session.refresh(db_user)
+    return model.Token(access_token=auth_handler.create_access_token(db_user), token_type="bearer")

app/auth/auth_handler.py

@@ -0,0 +1,59 @@
+import time
+from typing import Dict, Annotated
+from datetime import datetime, timedelta, timezone
+from typing import Union
+
+import jwt
+from decouple import config
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+from fastapi import Depends, HTTPException, status
+
+import app.model as model
+import app.auth.crypto as crypto
+
+
+JWT_SECRET = "96706f2fb5daf56405a9f2554399e53e93fe2b5f22c1b6bfc48c1d1f77a79151"
+JWT_ALGORITHM = "HS256"
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+def check_and_get_user(data: model.UserLogin, session: model.Session):
+    user = session.query(model.User).filter(model.User.email == data.email).first()
+    if user:
+        if crypto.verify_password(data.password, user.password):
+            return user
+    return None
+
+def create_access_token(user: model.User, expires_delta: Union[timedelta, None] = None):
+    to_encode = {
+        "sub": {'id': user.id, 'email': user.email}
+    }
+    if expires_delta:
+        expire = datetime.now(timezone.utc) + expires_delta
+    else:
+        expire = datetime.now(timezone.utc) + timedelta(minutes=30)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, JWT_SECRET, algorithm=JWT_ALGORITHM)
+    return encoded_jwt
+
+def decode_jwt(token: str) -> dict:
+    try:
+        decoded_token = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
+        return decoded_token
+    except:
+        return None
+
+async def get_current_user_id(token: Annotated[str, Depends(oauth2_scheme)]):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        decoded_token = decode_jwt(token)
+        if not decoded_token:
+            raise credentials_exception
+        user_id = model.UserId(**decoded_token['sub'])
+        return user_id
+    except:
+        raise credentials_exception

app/auth/crypto.py

@@ -0,0 +1,32 @@
+from passlib.context import CryptContext
+import base64
+import os
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+def verify_password(plain_password: str, hashed_password: str):
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+def get_password_hash(password: str):
+    return pwd_context.hash(password)
+
+
+def generate_random_base64_string(length: int):
+  """Generates a random string of the specified length and encodes it in base64 (URL-safe).
+
+  Args:
+      length: The desired length of the random string (in bytes).
+
+  Returns:
+      A string containing the random data encoded in base64 (URL-safe).
+  """
+  # Generate random bytes using os.urandom()
+  random_bytes = os.urandom(length)
+
+  # Encode the random bytes in base64 (URL-safe) format
+  encoded_string = base64.urlsafe_b64encode(random_bytes).decode()
+
+  # Remove trailing newline character (optional)
+  return encoded_string.rstrip("\n")

app/model.py

@@ -0,0 +1,51 @@
+from pydantic import BaseModel, Field, EmailStr
+from typing import Optional, Union
+
+from sqlmodel import Field, Session, SQLModel, Relationship, create_engine
+
+sqlite_file_name = "database.db"
+sqlite_url = f"sqlite:///{sqlite_file_name}"
+
+connect_args = {"check_same_thread": False}
+engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)
+
+def get_db_session():
+    with Session(engine) as session:
+        yield session
+
+def create_db_and_tables():
+    SQLModel.metadata.create_all(engine)
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+class TokenData(BaseModel):
+    username: Union[str, None] = None
+
+class UserLogin(SQLModel):
+    email: EmailStr = Field(...)
+    password: str = Field(...)
+
+class UserId(SQLModel):
+    id: int
+    email: EmailStr
+
+class UserCreate(UserLogin):
+    fullname: Optional[str] = Field(...)
+
+class User(UserCreate, table=True):
+    id: int = Field(default=None, primary_key=True)
+    #screenshots: list["Screenshot"] = Relationship(back_populates="owner")
+
+class ScreenshotCreate(SQLModel):
+    url: str = Field(...)
+    img: str = Field(...)
+
+class ScreenshotBase(ScreenshotCreate):
+    owner_id: int | None = Field(default=None, foreign_key="user.id")
+    external_id: str = Field(default = None)
+    #owner: User | None = Relationship(back_populates="screenshots")
+
+class Screenshot(ScreenshotBase, table=True):
+    id: int = Field(default=None, primary_key=True)

main.py

@@ -0,0 +1,4 @@
+import uvicorn
+
+if __name__ == "__main__":
+    uvicorn.run("app.api:app_obj", host="0.0.0.0", port=8081, reload=True)

requirements.txt

@@ -0,0 +1,8 @@
+email_validator==2.1.1
+fastapi==0.111.0
+PyJWT==2.8.0
+python-decouple==3.8
+uvicorn==0.29.0
+sqlmodel==0.0.19
+passlib==1.7.4
+bcrypt==4.1.3