Merge pull request #70 from Utkarsh4517/feat-sdk

Feat: Created SDK

sdk/example.py

@@ -0,0 +1,27 @@
+from sdk.module import PasteBinSDK
+
+def test_pastebin_sdk():
+    sdk = PasteBinSDK()
+
+    try:
+        # Create a paste
+        paste_id = sdk.create_paste("print('Hello, World!')", ".py")
+        print(f"Created paste with ID: {paste_id}")
+
+        # Retrieve the paste
+        content = sdk.get_paste(paste_id)
+        print(f"Retrieved paste content: {content}")
+
+        # Delete the paste
+        result = sdk.delete_paste(paste_id)
+        print(f"Delete result: {result}")
+
+        # Get supported languages
+        languages = sdk.get_languages()
+        print(f"Number of supported languages: {len(languages)}")
+
+    except RuntimeError as e:
+        print(f"An error occurred: {e}")
+
+if __name__ == "__main__":
+    test_pastebin_sdk()

sdk/sdk/module.py

@@ -0,0 +1,68 @@
+import requests
+from typing import Optional, Union
+from pathlib import Path
+
+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:
+        """
+        Create a new paste.
+        :param content: The content to paste, either as a string or a Path to a file
+        :param file_extension: File extension for syntax highlighting (required)
+        :return: The unique identifier of the created paste
+        """
+        try:
+            if isinstance(content, Path):
+                with open(content, 'r', encoding='utf-8') as f:
+                    content = f.read()
+
+            data = {
+                'content': content,
+                'extension': file_extension
+            }
+            response = requests.post(f"{self.base_url}/api/paste", json=data)
+            response.raise_for_status()
+            result = response.json()
+            return result['uuid']
+        except requests.RequestException as e:
+            raise RuntimeError(f"Error creating paste: {str(e)}")
+
+    def get_paste(self, uuid: str) -> dict:
+        """
+        Retrieve a paste by its unique identifier.
+        :param uuid: The unique identifier of the paste
+        :return: A dictionary containing the paste details (uuid, content, extension)
+        """
+        try:
+            response = requests.get(f"{self.base_url}/api/paste/{uuid}")
+            response.raise_for_status()
+            return response.json()
+        except requests.RequestException as e:
+            raise RuntimeError(f"Error retrieving paste: {str(e)}")
+
+    def delete_paste(self, uuid: str) -> str:
+        """
+        Delete a paste by its unique identifier.
+        :param uuid: The unique identifier of the paste
+        :return: A confirmation message
+        """
+        try:
+            response = requests.delete(f"{self.base_url}/paste/{uuid}")
+            response.raise_for_status()
+            return response.text
+        except requests.RequestException as e:
+            raise RuntimeError(f"Error deleting paste: {str(e)}")
+
+    def get_languages(self) -> dict:
+        """
+        Get the list of supported languages for syntax highlighting.
+        :return: A dictionary of supported languages
+        """
+        try:
+            response = requests.get(f"{self.base_url}/languages.json")
+            response.raise_for_status()
+            return response.json()
+        except requests.RequestException as e:
+            raise RuntimeError(f"Error fetching languages: {str(e)}")

src/paste/main.py

@@ -15,6 +15,9 @@
     RedirectResponse,
     JSONResponse,
 )
+from starlette.requests import Request
+from starlette.responses import Response
+from typing import Callable, Awaitable, List, Optional, Union, Any
 import shutil
 import os
 import json
@@ -30,8 +33,8 @@
 from pygments.lexers import get_lexer_by_name, guess_lexer
 from pygments.formatters import HtmlFormatter
 from pygments.util import ClassNotFound
-from typing import List, Optional
 from . import __version__, __author__, __contact__, __url__
+from .schema import PasteCreate, PasteResponse, PasteDetails
 
 description: str = "paste.py 🐍 - A pastebin written in python."
 
@@ -50,7 +53,20 @@
     redoc_url=None,
 )
 app.state.limiter = limiter
-app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
+
+
+def rate_limit_exceeded_handler(request: Request, exc: Exception) -> Union[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
+    )
+
+app.add_exception_handler(RateLimitExceeded, rate_limit_exceeded_handler)
 
 origins: List[str] = ["*"]
 
@@ -306,3 +322,64 @@ async def get_languages() -> JSONResponse:
             detail=f"Error reading languages file: {e}",
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
         )
+
+# apis to create and get a paste which returns uuid and url (to be used by SDK)
+@app.post("/api/paste", response_model=PasteResponse)
+async def create_paste(paste: PasteCreate) -> JSONResponse:
+    try:
+        uuid: str = generate_uuid()
+        if uuid in large_uuid_storage:
+            uuid = generate_uuid()
+
+        uuid_with_extension: str = f"{uuid}.{paste.extension}"
+        path: str = f"data/{uuid_with_extension}"
+
+        with open(path, "w", encoding="utf-8") as f:
+            f.write(paste.content)
+
+        large_uuid_storage.append(uuid_with_extension)
+
+        return JSONResponse(
+            content=PasteResponse(
+                uuid=uuid_with_extension,
+                url=f"{BASE_URL}/paste/{uuid_with_extension}"
+            ).dict(),
+            status_code=status.HTTP_201_CREATED
+        )
+    except Exception as e:
+        raise HTTPException(
+            detail=f"There was an error creating the paste: {str(e)}",
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+        )
+
+@app.get("/api/paste/{uuid}", response_model=PasteDetails)
+async def get_paste_details(uuid: str) -> JSONResponse:
+    if not "." in uuid:
+        uuid = _find_without_extension(uuid)
+    path: str = f"data/{uuid}"
+
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            content: str = f.read()
+
+        extension: str = Path(path).suffix[1:]
+
+        return JSONResponse(
+            content=PasteDetails(
+                uuid=uuid,
+                content=content,
+                extension=extension
+            ).dict(),
+            status_code=status.HTTP_200_OK
+        )
+    except FileNotFoundError:
+        raise HTTPException(
+            detail="Paste not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+        )
+    except Exception as e:
+        raise HTTPException(
+            detail=f"Error retrieving paste: {str(e)}",
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+        )
+

src/paste/schema.py

@@ -1,5 +1,18 @@
+from typing import Optional
 from pydantic import BaseModel
 
-
 class Data(BaseModel):
-    input_data: str
+    input_data: str
+
+class PasteCreate(BaseModel):
+    content: str
+    extension: Optional[str] = None
+
+class PasteResponse(BaseModel):
+    uuid: str
+    url: str
+
+class PasteDetails(BaseModel):
+    uuid: str
+    content: str
+    extension: Optional[str] = None

tests/test_api.py

@@ -90,3 +90,48 @@ def test_post_file_route_size_limit() -> None:
     os.remove(large_file_name)
     assert response.status_code == 413
     assert "File is too large" in response.text
+
+def test_post_api_paste_route() -> None:
+    paste_data = {
+        "content": "This is a test paste content",
+        "extension": "txt"
+    }
+    response = client.post("/api/paste", json=paste_data)
+    assert response.status_code == 201
+    response_json = response.json()
+    assert "uuid" in response_json
+    assert "url" in response_json
+    assert response_json["uuid"].endswith(".txt")
+    assert response_json["url"].startswith("http://paste.fosscu.org/paste/")
+
+    # Clean up: delete the created paste
+    uuid = response_json["uuid"]
+    delete_response = client.delete(f"/paste/{uuid}")
+    assert delete_response.status_code == 200
+
+def test_get_api_paste_route() -> None:
+    # First, create a paste
+    paste_data = {
+        "content": "This is a test paste content for GET",
+        "extension": "md"
+    }
+    create_response = client.post("/api/paste", json=paste_data)
+    assert create_response.status_code == 201
+    created_uuid = create_response.json()["uuid"]
+
+    # Now, test getting the paste
+    response = client.get(f"/api/paste/{created_uuid}")
+    assert response.status_code == 200
+    response_json = response.json()
+    assert response_json["uuid"] == created_uuid
+    assert response_json["content"] == paste_data["content"]
+    assert response_json["extension"] == paste_data["extension"]
+
+    # Clean up: delete the created paste
+    delete_response = client.delete(f"/paste/{created_uuid}")
+    assert delete_response.status_code == 200
+
+def test_get_api_paste_route_not_found() -> None:
+    response = client.get("/api/paste/nonexistent_uuid.txt")
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Paste not found"