Skip to content
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 .pre-commit-config.yaml
Original file line number Diff line number Diff line change
@@ -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:
Expand Down
6 changes: 3 additions & 3 deletions alembic/env.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
13 changes: 7 additions & 6 deletions alembic/versions/9513acd42747_initial_migration.py
Original file line number Diff line number Diff line change
@@ -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:
Expand Down
14 changes: 7 additions & 7 deletions sdk/sdk/module.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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:
"""
Expand All @@ -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:
"""
Expand All @@ -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:
"""
Expand All @@ -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}")
8 changes: 4 additions & 4 deletions src/paste/logging.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from typing import Any, Dict
from typing import Any

from pydantic import BaseModel

Expand All @@ -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},
}
33 changes: 17 additions & 16 deletions src/paste/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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)
Expand All @@ -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

Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down
5 changes: 2 additions & 3 deletions src/paste/minio.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
import io
import uuid
from typing import Optional

import boto3
from botocore.client import Config
Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand Down
8 changes: 4 additions & 4 deletions src/paste/schema.py
Original file line number Diff line number Diff line change
@@ -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

Expand All @@ -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):
Expand All @@ -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):
Expand Down
4 changes: 2 additions & 2 deletions src/paste/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
import re
import string
from pathlib import Path
from typing import Pattern
from re import Pattern


def generate_uuid() -> str:
Expand Down Expand Up @@ -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]

Expand Down
Loading