mirror of
https://github.com/Xevion/linkpulse.git
synced 2025-12-06 05:15:35 -06:00
Minor formatting concerns, remove dangling IPAddress usage
This commit is contained in:
9
.vscode/settings.json
vendored
9
.vscode/settings.json
vendored
@@ -4,19 +4,26 @@
|
||||
"backref",
|
||||
"bpython",
|
||||
"Callsite",
|
||||
"clsx",
|
||||
"excepthook",
|
||||
"inmemory",
|
||||
"linkpulse",
|
||||
"migratehistory",
|
||||
"Nixpacks",
|
||||
"ORJSON",
|
||||
"pext",
|
||||
"pwdlib",
|
||||
"pytest",
|
||||
"pytz",
|
||||
"rtype",
|
||||
"starlette",
|
||||
"structlog",
|
||||
"tailwindcss",
|
||||
"timestamper"
|
||||
],
|
||||
"python.analysis.extraPaths": ["./backend/"],
|
||||
"[github-actions-workflow]": {
|
||||
"editor.formatOnSave": false
|
||||
}
|
||||
},
|
||||
"python.analysis.diagnosticMode": "workspace"
|
||||
}
|
||||
|
||||
@@ -16,17 +16,18 @@ setup_logging()
|
||||
|
||||
import os
|
||||
import sys
|
||||
import structlog
|
||||
|
||||
import structlog
|
||||
|
||||
logger = structlog.get_logger()
|
||||
|
||||
|
||||
def main(*args):
|
||||
"""
|
||||
Primary entrypoint for the LinkPulse application
|
||||
- Don't import any modules globally unless you're certain it's necessary. Imports should be tightly controlled.
|
||||
"""
|
||||
def main(*args: str) -> None:
|
||||
"""Primary entrypoint for the LinkPulse application
|
||||
NOTE: Don't import any modules globally unless you're certain it's necessary. Imports should be tightly controlled.
|
||||
:param args: The command-line arguments to parse and execute.
|
||||
:type args: str"""
|
||||
|
||||
if args[0] == "serve":
|
||||
from linkpulse.utilities import is_development
|
||||
from uvicorn import run
|
||||
@@ -57,9 +58,9 @@ def main(*args):
|
||||
|
||||
# import most useful objects, models, and functions
|
||||
lp = linkpulse # alias
|
||||
from linkpulse.utilities import get_db
|
||||
from linkpulse.app import app
|
||||
from linkpulse.models import BaseModel, IPAddress
|
||||
from linkpulse.models import BaseModel, User, Session
|
||||
from linkpulse.utilities import get_db
|
||||
|
||||
db = get_db()
|
||||
|
||||
|
||||
@@ -36,7 +36,7 @@ with suppress(ImportError):
|
||||
|
||||
def migrate(migrator: Migrator, database: pw.Database, *, fake=False):
|
||||
"""Write your migrations here."""
|
||||
|
||||
|
||||
@migrator.create_model
|
||||
class User(pw.Model):
|
||||
id = pw.AutoField()
|
||||
@@ -48,12 +48,12 @@ def migrate(migrator: Migrator, database: pw.Database, *, fake=False):
|
||||
class Meta:
|
||||
table_name = "user"
|
||||
|
||||
migrator.remove_model('ipaddress')
|
||||
migrator.remove_model("ipaddress")
|
||||
|
||||
|
||||
def rollback(migrator: Migrator, database: pw.Database, *, fake=False):
|
||||
"""Write your rollback migrations here."""
|
||||
|
||||
|
||||
@migrator.create_model
|
||||
class IPAddress(pw.Model):
|
||||
ip = pw.CharField(max_length=255, primary_key=True)
|
||||
@@ -63,4 +63,4 @@ def rollback(migrator: Migrator, database: pw.Database, *, fake=False):
|
||||
class Meta:
|
||||
table_name = "ipaddress"
|
||||
|
||||
migrator.remove_model('user')
|
||||
migrator.remove_model("user")
|
||||
|
||||
@@ -9,12 +9,21 @@ router = APIRouter()
|
||||
def validate_session(
|
||||
token: str, user: bool = True
|
||||
) -> Tuple[bool, bool, Optional[User]]:
|
||||
"""
|
||||
Given a token, validate that the session exists and is not expired.
|
||||
"""Given a token, validate that the session exists and is not expired.
|
||||
|
||||
This function has side effects:
|
||||
- This function updates last_used if `user` is True.
|
||||
- This function will invalidate the session if it is expired.
|
||||
|
||||
:param token: The session token to validate.
|
||||
:type token: str
|
||||
:param user: Whether to update the last_used timestamp of the session.
|
||||
:type user: bool
|
||||
:return: A tuple containing:
|
||||
- A boolean indicating if the session exists.
|
||||
- A boolean indicating if the session is valid.
|
||||
- The User object if the session is valid, otherwise None.
|
||||
:rtype: Tuple[bool, bool, Optional[User]]
|
||||
"""
|
||||
# Check if session exists
|
||||
session = Session.get_or_none(Session.token == token)
|
||||
@@ -28,3 +37,46 @@ def validate_session(
|
||||
if user:
|
||||
session.use()
|
||||
return True, True, session.user
|
||||
|
||||
|
||||
@router.post("/api/login")
|
||||
async def login():
|
||||
# Validate parameters
|
||||
# Hash regardless of user existence to prevent timing attacks
|
||||
# Check if user exists, return 401 if not
|
||||
# Check if password matches, return 401 if not
|
||||
# Create session
|
||||
# Set Cookie of session token
|
||||
# Return 200 with mild user information
|
||||
pass
|
||||
|
||||
|
||||
@router.post("/api/logout")
|
||||
async def logout():
|
||||
# TODO: Force logout parameter, logout ALL sessions for User
|
||||
# Get session token from Cookie
|
||||
# Delete session
|
||||
# Return 200
|
||||
pass
|
||||
|
||||
|
||||
@router.post("/api/register")
|
||||
async def register():
|
||||
# Validate parameters
|
||||
# Hash password
|
||||
# Create User
|
||||
# Create Session
|
||||
# Set Cookie of session token
|
||||
# Return 200 with mild user information
|
||||
pass
|
||||
|
||||
|
||||
@router.get("/api/sessions")
|
||||
async def sessions():
|
||||
pass
|
||||
|
||||
|
||||
# GET /api/user/{id}/sessions
|
||||
# GET /api/user/{id}/sessions/{token}
|
||||
# DELETE /api/user/{id}/sessions
|
||||
# POST /api/user/{id}/logout (delete all sessions)
|
||||
|
||||
@@ -1,3 +1,7 @@
|
||||
"""Miscellaneous endpoints for the Linkpulse API."""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter
|
||||
from fastapi_cache.decorator import cache
|
||||
from linkpulse.utilities import get_db
|
||||
@@ -9,15 +13,19 @@ db = get_db()
|
||||
|
||||
@router.get("/health")
|
||||
async def health():
|
||||
"""An endpoint to check if the service is running.
|
||||
:return: OK
|
||||
:rtype: Literal['OK']"""
|
||||
# TODO: Check database connection
|
||||
return "OK"
|
||||
|
||||
|
||||
@router.get("/api/migration")
|
||||
@cache(expire=60)
|
||||
async def get_migration():
|
||||
"""
|
||||
Returns the details of the most recent migration.
|
||||
async def get_migration() -> dict[str, Any]:
|
||||
"""Get the last migration name and timestamp from the migratehistory table.
|
||||
:return: The last migration name and timestamp.
|
||||
:rtype: dict[str, Any]
|
||||
"""
|
||||
# Kind of insecure, but this is just a demo thing to show that migratehistory is available.
|
||||
cursor = db.execute_sql(
|
||||
|
||||
Reference in New Issue
Block a user