chore: formatting

ci (pyproject.toml): enable preview rules

Exposed more static analysis from ruff in editor

chore (auth.py): formatting

Author: pythoninthegrass

app/core/auth.py

@@ -38,6 +38,7 @@
     auto_error=False
 )
 
+
 class CookieOrHeaderToken(SecurityBase):
     def __init__(
         self,
@@ -73,21 +74,26 @@ async def __call__(self, request: Request) -> str | None:
 
         return token
 
+
 # Security schemes
 security = CookieOrHeaderToken(auto_error=False)  # Don't auto-error so we can try form auth
 oauth_form_dependency = Depends(OAuth2PasswordRequestForm)
 
+
 class IPConfig(BaseModel):
     whitelist: list[str] = ["localhost", "127.0.0.1"]
     public_ips: list[str] = []  # TODO: add whitelisted public IPs here
 
+
 ip_config = IPConfig()
 
+
 def is_ip_allowed(request: Request):
     """Check if the client IP is in the whitelist"""
     client_host = request.client.host
     return client_host in ip_config.whitelist or client_host in ip_config.public_ips
 
+
 async def get_current_user(
     request: Request,
     token: str | None = Depends(security),
@@ -120,18 +126,21 @@ async def get_current_user(
 
     return user
 
+
 async def get_current_active_user(current_user: User = Depends(get_current_user)):
     """Get current active user"""
     if current_user.disabled:
         raise HTTPException(status_code=400, detail="Inactive user")
     return current_user
 
+
 async def ip_whitelist_or_auth(request: Request, current_user: User = Depends(get_current_active_user)):
     """Check if request is from whitelisted IP or authenticated user"""
     if is_ip_allowed(request):
         return {"bypass_auth": True}
     return current_user
 
+
 def check_auth(auth: dict | User) -> None:
     """
     Shared function to check authentication result.
@@ -142,6 +151,7 @@ def check_auth(auth: dict | User) -> None:
     elif not isinstance(auth, User):
         raise HTTPException(status_code=401, detail="Unauthorized")
 
+
 # Dependencies
 current_user_dependency = Depends(get_current_user)
 current_active_user_dependency = Depends(get_current_active_user)

app/core/db.py

@@ -21,12 +21,14 @@
 # Initialize database
 db = Database()
 
+
 # User model
 class UserInfo(db.Entity):
     username = Required(str, unique=True)
     hashed_password = Required(str)
     email = Optional(str)
 
+
 # Initialize database connection
 def init_db():
     db.bind(provider='postgres',
@@ -44,33 +46,41 @@ def init_db():
             hashed_password = pwd_context.hash(DB_PASS)
             UserInfo(username=DB_USER, hashed_password=hashed_password)
 
+
 # Authentication models
 class Token(BaseModel):
     access_token: str
     token_type: str
 
+
 class TokenData(BaseModel):
     username: str | None = None
 
+
 class User(BaseModel):
     username: str
     email: str | None = None
     disabled: bool | None = None
 
+
 class UserInDB(User):
     hashed_password: str
 
+
 # Password handling
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
+
 def verify_password(plain_password, hashed_password):
     """Validate plaintext password against hashed password"""
     return pwd_context.verify(plain_password, hashed_password)
 
+
 def get_password_hash(password):
     """Return hashed password"""
     return pwd_context.hash(password)
 
+
 # User management
 def get_user(username: str):
     """Get user from database"""
@@ -79,6 +89,7 @@ def get_user(username: str):
         if user:
             return UserInDB(username=user.username, hashed_password=user.hashed_password)
 
+
 def authenticate_user(username: str, password: str):
     """Authenticate user"""
     user = get_user(username)
@@ -88,6 +99,7 @@ def authenticate_user(username: str, password: str):
         return False
     return user
 
+
 def load_user(username: str):
     """Load user from database"""
     with db_session:
@@ -97,6 +109,7 @@ def load_user(username: str):
         else:
             raise HTTPException(status_code=404, detail="User not found")
 
+
 # Token management
 def create_access_token(data: dict, expires_delta: timedelta | None = None):
     """Create access token"""

app/core/slackbot.py

@@ -32,6 +32,7 @@
 # strip double quotes from env strings (local image)
 CHANNEL = CHANNEL.strip('"')
 
+
 def load_channels():
     # read channel
     chan = pd.read_csv('channels.csv', header=0)
@@ -64,6 +65,7 @@ def load_channels():
 
     return channels
 
+
 # python sdk
 client = WebClient(token=BOT_USER_TOKEN)
 

app/main.py

@@ -102,6 +102,7 @@
     "clientId": "meetup_bot",
 }
 
+
 @app.post("/token", response_model=Token)
 async def login_for_oauth_token(response: Response, form_data: OAuth2PasswordRequestForm = oauth_form_dependency):
     """Login for oauth access token"""
@@ -130,16 +131,19 @@ async def login_for_oauth_token(response: Response, form_data: OAuth2PasswordReq
 
     return {"access_token": oauth_token, "token_type": "bearer"}
 
+
 @app.get("/healthz", status_code=200)
 def health_check():
     """Smoke test to check if the app is running"""
     return {"status": "ok"}
 
+
 @app.get("/", response_class=HTMLResponse)
 def index(request: Request):
     with open(Path("resources/templates/login.html")) as f:
         return HTMLResponse(content=f.read(), status_code=200)
 
+
 @app.post("/auth/login")
 def login(request: Request, username: str = Form(...), password: str = Form(...)):
     """Redirect to "/docs" from index page if user successfully logs in with HTML form"""
@@ -169,6 +173,7 @@ def login(request: Request, username: str = Form(...), password: str = Form(...)
         detail="Incorrect username or password"
     )
 
+
 @api_router.get("/token")
 def generate_token(current_user: User = current_active_user_dependency):
     """
@@ -195,6 +200,7 @@ def generate_token(current_user: User = current_active_user_dependency):
 
     return access_token, refresh_token
 
+
 @api_router.get("/events")
 def get_events(auth: dict = ip_whitelist_auth_dependency,
                location: str = "Oklahoma City",
@@ -225,6 +231,7 @@ def get_events(auth: dict = ip_whitelist_auth_dependency,
 
     return events
 
+
 @api_router.get("/check-schedule")
 def should_post_to_slack(auth: dict = ip_whitelist_auth_dependency, request: Request = None):
     """
@@ -267,6 +274,7 @@ def should_post_to_slack(auth: dict = ip_whitelist_auth_dependency, request: Req
                 "should_post": False,
             }
 
+
 @api_router.post("/slack")
 def post_slack(
     auth: dict = ip_whitelist_auth_dependency,
@@ -321,6 +329,7 @@ def post_slack(
     except Exception as e:
         return {"message": f"Error posting to Slack: {str(e)}", "status": "error"}
 
+
 @api_router.post("/snooze")
 def snooze_slack_post(
     duration: str,
@@ -350,6 +359,7 @@ def snooze_slack_post(
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e)) from e
 
+
 @api_router.get("/schedule")
 def get_current_schedule(auth: dict | User = ip_whitelist_auth_dependency):
     """
@@ -375,6 +385,7 @@ def get_current_schedule(auth: dict | User = ip_whitelist_auth_dependency):
 
     return {"schedules": schedules}
 
+
 # routes
 app.include_router(api_router)
 

pyproject.toml

@@ -148,6 +148,9 @@ select = [
 ]
 ignore = ["D203", "E203", "E251", "E266", "E401", "E402", "E501", "F401", "F403", "F405"]
 
+# enable preview style formatting
+preview = true
+
 # Allow unused variables when underscore-prefixed.
 dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"