diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 6f4b854..13552b0 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,6 +1,6 @@ repos: - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.1.9 + rev: v0.16.0 hooks: - id: ruff args: diff --git a/alembic/env.py b/alembic/env.py index 15b6d68..706719f 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -14,11 +14,11 @@ sys.path.insert(0, src_dir) # Import your models and Base after path setup -from paste.config import get_settings # noqa: E402 -from paste.database import Base # noqa: E402 +from paste.config import get_settings +from paste.database import Base # Import all your models here so Base.metadata is populated -from paste.models import Paste # noqa: F401, E402 +from paste.models import Paste # noqa: F401 # this is the Alembic Config object config = context.config diff --git a/alembic/versions/9513acd42747_initial_migration.py b/alembic/versions/9513acd42747_initial_migration.py index dee74fb..1144b39 100644 --- a/alembic/versions/9513acd42747_initial_migration.py +++ b/alembic/versions/9513acd42747_initial_migration.py @@ -1,21 +1,22 @@ """Initial migration Revision ID: 9513acd42747 -Revises: +Revises: Create Date: 2025-02-09 02:54:48.803960 """ -from typing import Sequence, Union -from alembic import op +from collections.abc import Sequence + import sqlalchemy as sa +from alembic import op # revision identifiers, used by Alembic. revision: str = "9513acd42747" -down_revision: Union[str, None] = None -branch_labels: Union[str, Sequence[str], None] = None -depends_on: Union[str, Sequence[str], None] = None +down_revision: str | None = None +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None def upgrade() -> None: diff --git a/sdk/sdk/module.py b/sdk/sdk/module.py index 585704e..22cee77 100644 --- a/sdk/sdk/module.py +++ b/sdk/sdk/module.py @@ -1,13 +1,13 @@ -import requests -from typing import Union from pathlib import Path +import requests + class PasteBinSDK: def __init__(self, base_url: str = "https://paste.fosscu.org"): self.base_url = base_url - def create_paste(self, content: Union[str, Path], file_extension: str) -> str: + def create_paste(self, content: str | Path, file_extension: str) -> str: """ Create a new paste. :param content: The content to paste, either as a string or a Path to a file @@ -25,7 +25,7 @@ def create_paste(self, content: Union[str, Path], file_extension: str) -> str: result = response.json() return result["uuid"] except requests.RequestException as e: - raise RuntimeError(f"Error creating paste: {str(e)}") + raise RuntimeError(f"Error creating paste: {e!s}") def get_paste(self, uuid: str) -> dict: """ @@ -38,7 +38,7 @@ def get_paste(self, uuid: str) -> dict: response.raise_for_status() return response.json() except requests.RequestException as e: - raise RuntimeError(f"Error retrieving paste: {str(e)}") + raise RuntimeError(f"Error retrieving paste: {e!s}") def delete_paste(self, uuid: str) -> str: """ @@ -51,7 +51,7 @@ def delete_paste(self, uuid: str) -> str: response.raise_for_status() return response.text except requests.RequestException as e: - raise RuntimeError(f"Error deleting paste: {str(e)}") + raise RuntimeError(f"Error deleting paste: {e!s}") def get_languages(self) -> dict: """ @@ -63,4 +63,4 @@ def get_languages(self) -> dict: response.raise_for_status() return response.json() except requests.RequestException as e: - raise RuntimeError(f"Error fetching languages: {str(e)}") + raise RuntimeError(f"Error fetching languages: {e!s}") diff --git a/src/paste/logging.py b/src/paste/logging.py index 3f94793..d43a4b4 100644 --- a/src/paste/logging.py +++ b/src/paste/logging.py @@ -1,4 +1,4 @@ -from typing import Any, Dict +from typing import Any from pydantic import BaseModel @@ -13,20 +13,20 @@ class LogConfig(BaseModel): # Logging config version: int = 1 disable_existing_loggers: bool = False - formatters: Dict[str, Dict[str, str]] = { + formatters: dict[str, dict[str, str]] = { "default": { "()": "uvicorn.logging.DefaultFormatter", "fmt": LOG_FORMAT, "datefmt": "%Y-%m-%d %H:%M:%S", }, } - handlers: Dict[str, Dict[str, Any]] = { + handlers: dict[str, dict[str, Any]] = { "default": { "formatter": "default", "class": "logging.StreamHandler", "stream": "ext://sys.stderr", }, } - loggers: Dict[str, Dict[str, Any]] = { + loggers: dict[str, dict[str, Any]] = { LOGGER_NAME: {"handlers": ["default"], "level": LOG_LEVEL}, } diff --git a/src/paste/main.py b/src/paste/main.py index d14702b..fe95bfa 100644 --- a/src/paste/main.py +++ b/src/paste/main.py @@ -2,10 +2,11 @@ import json import logging import time +from collections.abc import Awaitable from datetime import datetime, timedelta, timezone from logging.config import dictConfig from pathlib import Path -from typing import Awaitable, List, Optional, Union, cast +from typing import cast from fastapi import Depends, FastAPI, File, Form, Header, HTTPException, Query, Request, Response, UploadFile, status from fastapi.middleware.cors import CORSMiddleware @@ -47,14 +48,14 @@ async def delete_expired_urls() -> None: while True: - db: Optional[Session] = None + db: Session | None = None try: db = Session_Local() current_time = datetime.utcnow() # Find and delete expired URLs - expired_urls: List[Paste] = db.query(Paste).filter(Paste.expiresat <= current_time).all() + expired_urls: list[Paste] = db.query(Paste).filter(Paste.expiresat <= current_time).all() for url in expired_urls: if url.s3_link is not None: @@ -96,7 +97,7 @@ async def delete_expired_urls() -> None: app.state.limiter = limiter -def rate_limit_exceeded_handler(request: Request, exc: Exception) -> Union[Response, Awaitable[Response]]: +def rate_limit_exceeded_handler(request: Request, exc: Exception) -> Response | Awaitable[Response]: if isinstance(exc, RateLimitExceeded): return Response(content="Rate limit exceeded", status_code=429) return Response(content="An error occurred", status_code=500) @@ -136,7 +137,7 @@ async def startup_event(): create_bucket_if_not_exists() -origins: List[str] = ["*"] +origins: list[str] = ["*"] BASE_URL: str = get_settings().BASE_URL app.add_middleware( @@ -208,16 +209,16 @@ async def health(db: Session = Depends(get_db)) -> HealthResponse: async def get_paste_data( request: Request, uuid: str, - user_agent: Optional[str] = Header(None), + user_agent: str | None = Header(None), db: Session = Depends(get_db), ) -> Response: try: uuid = extract_uuid(uuid) - data: Optional[Paste] = db.query(Paste).filter(Paste.pasteID == uuid).first() + data: Paste | None = db.query(Paste).filter(Paste.pasteID == uuid).first() - content: Optional[str] = None - extension: Optional[str] = None + content: str | None = None + extension: str | None = None if data is None: raise HTTPException(detail="Paste not found", status_code=status.HTTP_404_NOT_FOUND) @@ -235,7 +236,7 @@ async def get_paste_data( if extension is None: extension = "" - extension = extension[1::] if extension.startswith(".") else extension + extension = extension.removeprefix(".") is_browser_request = "Mozilla" in user_agent if user_agent else False @@ -288,11 +289,11 @@ async def get_paste_data( async def post_as_a_file( request: Request, file: UploadFile = File(...), - expiration: Optional[str] = Query(None, description="Expiration time: '1h', '1d', '1w', '1m', or ISO datetime"), + expiration: str | None = Query(None, description="Expiration time: '1h', '1d', '1w', '1m', or ISO datetime"), db: Session = Depends(get_db), ) -> PlainTextResponse: try: - file_extension: Optional[str] = None + file_extension: str | None = None # Extract file extension from the filename try: if file.filename is not None: @@ -409,9 +410,9 @@ async def web(request: Request) -> Response: async def web_post( request: Request, content: str = Form(...), - extension: Optional[str] = Form(None), - expiration: Optional[str] = Form(None), - custom_expiry: Optional[str] = Form(None), + extension: str | None = Form(None), + expiration: str | None = Form(None), + custom_expiry: str | None = Form(None), db: Session = Depends(get_db), ) -> RedirectResponse: try: @@ -461,7 +462,7 @@ async def web_post( except Exception as e: db.rollback() raise HTTPException( - detail=f"There was an error creating the paste: {str(e)}", + detail=f"There was an error creating the paste: {e!s}", status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, ) finally: diff --git a/src/paste/minio.py b/src/paste/minio.py index 82b004f..fd3bfbf 100644 --- a/src/paste/minio.py +++ b/src/paste/minio.py @@ -1,6 +1,5 @@ import io import uuid -from typing import Optional import boto3 from botocore.client import Config @@ -41,7 +40,7 @@ def get_object_data(object_name: str, bucket_name: str = get_settings().MINIO_BU def post_object_data( object_data: str, - object_name: Optional[str] = None, + object_name: str | None = None, bucket_name: str = get_settings().MINIO_BUCKET_NAME, ) -> str: try: @@ -66,7 +65,7 @@ def post_object_data( def post_object_data_as_file( source_file_path: str, - object_name: Optional[str] = None, + object_name: str | None = None, bucket_name: str = get_settings().MINIO_BUCKET_NAME, ) -> str: try: diff --git a/src/paste/schema.py b/src/paste/schema.py index 13817d3..ea55af2 100644 --- a/src/paste/schema.py +++ b/src/paste/schema.py @@ -1,6 +1,6 @@ import time from datetime import datetime -from typing import Literal, Optional, Union +from typing import Literal from pydantic import BaseModel, Field @@ -11,8 +11,8 @@ class Data(BaseModel): class PasteCreate(BaseModel): content: str - extension: Optional[str] = None - expiration: Optional[Union[Literal["1h", "1d", "1w", "1m"], datetime]] = None + extension: str | None = None + expiration: Literal["1h", "1d", "1w", "1m"] | datetime | None = None class PasteResponse(BaseModel): @@ -23,7 +23,7 @@ class PasteResponse(BaseModel): class PasteDetails(BaseModel): uuid: str content: str - extension: Optional[str] = None + extension: str | None = None class HealthResponse(BaseModel): diff --git a/src/paste/utils.py b/src/paste/utils.py index 8103d5b..e6e0a4a 100644 --- a/src/paste/utils.py +++ b/src/paste/utils.py @@ -3,7 +3,7 @@ import re import string from pathlib import Path -from typing import Pattern +from re import Pattern def generate_uuid() -> str: @@ -36,7 +36,7 @@ def _find_without_extension(file_name: str) -> str: pattern_without_dot: Pattern[str] = re.compile(r"^" + file_name + "$") math_pattern: list = [x for x in file_list if pattern_with_dot.match(x) or pattern_without_dot.match(x)] if len(math_pattern) == 0: - return str() + return "" else: return math_pattern[0]