Коммит f76431dc создал по автору vladzodchey's avatar vladzodchey ☯
Просмотр файлов

Implemented a netcat interface and moved db logic out of routes

владелец 858f8961
Скрыть пробелы
Построчно Рядом
.dockerignore
Просмотр файла @ f76431dc
!src/
!pyproject.toml
!uv.lock
!python-version
!.python-version
README.md
Просмотр файла @ f76431dc
......@@ -7,3 +7,4 @@ A lightweight selfhosted Pastebin-like app.
- [x] CRUD of Pastes.
- [ ] Web UI
- [x] Netcat (Termbin-like) behavior
src/api/pastes.py
Просмотр файла @ f76431dc
from typing import cast
from aiosqlite import IntegrityError, OperationalError
from fastapi import APIRouter, Depends, HTTPException, Response, Query
from fastapi.responses import JSONResponse
from http import HTTPStatus
from src.utils.auth import authorize
from src.utils.security import Permissions, generate_id
from src.schema import NewPaste, Paste, PastePreview, Page
from src.sqlite import database
MAX_RETRIES = 3
from src.utils.security import Permissions
from src.models import NewPaste
from src.schema import lookup_paste, modify_paste, page_pastes, push_paste, remove_paste
pastes = APIRouter(prefix="/pastes")
@pastes.get("/")
async def list_pastes(page: int = Query(1, ge=1), page_size: int = Query(10, ge=1, le=100), _ = Depends(authorize(Permissions.LIST))):
offset = (page - 1) * page_size
total = cast(tuple[int], await database.fetch_one("SELECT COUNT(*) FROM pastes", as_dict=False))[0]
rows = await database.fetch_all(
"""
SELECT id, title, sign, created_at, updated_at
FROM pastes
ORDER BY id
LIMIT ? OFFSET ?
""",
(page_size, offset)
)
items = [PastePreview(**row) for row in rows]
return Page[PastePreview](
items=items,
total=total,
page=page,
page_size=page_size
)
return await page_pastes(page, page_size)
@pastes.post("/")
async def create_paste(item: NewPaste, _ = Depends(authorize(Permissions.WRITE))):
while (retries := 0) < MAX_RETRIES:
try:
id = generate_id()
await database.execute(
"""
INSERT
INTO pastes (id, title, content, sign)
VALUES (?, ?, ?, ?)
""",
(id, item.title, item.content, item.sign)
)
return JSONResponse({"id": id}, HTTPStatus.CREATED)
except IntegrityError:
retries += 1
else:
raise HTTPException(
HTTPStatus.INTERNAL_SERVER_ERROR,
detail="Failed to generate an ID."
)
try:
id = await push_paste(item)
return JSONResponse({"id": id}, HTTPStatus.CREATED)
except RuntimeError as e:
raise HTTPException(HTTPStatus.CONFLICT, detail=str(e))
@pastes.get("/{id}")
async def get_paste(id: str, _ = Depends(authorize(Permissions.READ))):
c = await database.fetch_one(
"""
SELECT id, title, content, sign, created_at, updated_at
FROM pastes
WHERE id = ?
""",
(id,)
)
if c is None:
paste = await lookup_paste(id)
if paste is None:
raise HTTPException(HTTPStatus.NOT_FOUND, detail="Not found.")
return Paste(**dict(c))
return paste
@pastes.put("/{id}")
async def edit_paste(id: str, new_data: NewPaste, _ = Depends(authorize(Permissions.MODIFY))):
affected = await database.execute(
"""
UPDATE pastes
SET title = ?, content = ?, sign = ?, updated_at = datetime('now')
WHERE id = ?
""",
(new_data.title, new_data.content, new_data.sign, id)
)
if affected:
if await modify_paste(id, new_data):
return Response(status_code=HTTPStatus.NO_CONTENT)
else:
raise HTTPException(HTTPStatus.NOT_FOUND, detail="Not found.")
@pastes.delete("/{id}")
async def delete_paste(id: str, _ = Depends(authorize(Permissions.DELETE))):
affected = await database.execute(
"""
DELETE
FROM pastes
WHERE id = ?
""",
(id,)
)
if affected:
if await remove_paste(id):
return Response(status_code=HTTPStatus.NO_CONTENT)
else:
raise HTTPException(HTTPStatus.NOT_FOUND, detail="Not found.")
src/config.py
Просмотр файла @ f76431dc
......@@ -2,13 +2,14 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
app_env: str = "development"
environment: str = "development"
log_level: str = "info"
port: int = 8000
nc_enabled: bool = False
nc_port: int = 9999
db_url: str = "chancery.db"
permits: str = "read"
tokens: str | None = None
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
settings = Settings()
......
src/interfacing/netcat.py 0 → 100644
Просмотр файла @ f76431dc
import asyncio
from src.schema import push_paste
from src.models import NewPaste
async def save_and_generate_id(text: str) -> str:
return await push_paste(NewPaste(content=text))
async def handle_client(
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter
) -> None:
data = await reader.read(65536)
try:
text = data.decode('utf-8').rstrip('\r\n')
except UnicodeDecodeError:
writer.write(b"Error: Invalid UTF-8\n")
await writer.drain()
writer.close()
return
result_id = await save_and_generate_id(text)
response = f"{result_id}\n".encode()
writer.write(response)
await writer.drain()
writer.close()
await writer.wait_closed()
async def serve(host: str = "0.0.0.0", port: int = 9999):
server = await asyncio.start_server(
handle_client, host, port,
backlog=64,
reuse_port=True,
)
async with server:
await server.serve_forever()
src/main.py
Просмотр файла @ f76431dc
from fastapi import Depends, FastAPI, Response
from asyncio import create_task
from fastapi import Depends, FastAPI, Response, HTTPException
from http import HTTPStatus
from contextlib import asynccontextmanager
from inspect import signature
from src.config import get_settings
from src.config import get_settings, settings
from src.interfacing.netcat import serve
from src.schema import lookup_paste_raw
from src.sqlite import database
from src.api import api
from src.utils.auth import authorize
from src.utils.security import Permissions
@asynccontextmanager
async def lifespan(app: FastAPI):
await database.connect()
if settings.nc_enabled:
create_task(serve(port=settings.nc_port))
yield
await database.disconnect()
......@@ -25,3 +26,11 @@ app.include_router(api)
@app.get("/health")
def healthcheck(context=Depends(get_settings)):
return Response(status_code=HTTPStatus.OK)
@app.get("/raw/{id}")
async def get_paste_raw(id: str):
content = await lookup_paste_raw(id)
if content is None:
raise HTTPException(HTTPStatus.NOT_FOUND)
return Response(content)
src/models.py 0 → 100644
Просмотр файла @ f76431dc
from pydantic import BaseModel
from pydantic import PastDatetime
from typing import Generic, TypeVar
from datetime import datetime
T = TypeVar("T")
class PasteBase(BaseModel):
title: str | None = None
sign: str | None = None
class PasteID(BaseModel):
id: str
class PasteContent(BaseModel):
content: str
class Timestamps(BaseModel):
created_at: PastDatetime
updated_at: PastDatetime
class NewPaste(PasteBase, PasteContent):
pass
class PastePreview(PasteID, PasteBase, Timestamps):
pass
class Paste(PastePreview, PasteContent):
pass
class Page(BaseModel, Generic[T]):
items: list[T]
total: int
page: int
page_size: int
src/schema.py
Просмотр файла @ f76431dc
from pydantic import BaseModel
from pydantic import PastDatetime
from typing import Generic, TypeVar
from datetime import datetime
from aiosqlite import IntegrityError
from typing import cast
from src.sqlite import database
from src.models import NewPaste, Page, Paste, PastePreview
from src.utils.security import generate_id
T = TypeVar("T")
MAX_RETRIES = 3
class PasteBase(BaseModel):
title: str | None = None
sign: str | None = None
async def page_pastes(page: int, page_size: int) -> Page[PastePreview]:
offset = (page - 1) * page_size
total = cast(tuple[int], await database.fetch_one("SELECT COUNT(*) FROM pastes", as_dict=False))[0]
rows = await database.fetch_all(
"""
SELECT id, title, sign, created_at, updated_at
FROM pastes
ORDER BY created_at DESC
LIMIT ? OFFSET ?
""",
(page_size, offset)
)
items = [PastePreview(**row) for row in rows]
return Page[PastePreview](
items=items,
total=total,
page=page,
page_size=page_size
)
class PasteID(BaseModel):
id: str
class PasteContent(BaseModel):
content: str
async def push_paste(data: NewPaste) -> str:
while (retries := 0) <= MAX_RETRIES:
try:
new_id = generate_id()
await database.execute(
"""
INSERT
INTO pastes (id, title, content, sign)
VALUES (?, ?, ?, ?)
""",
(new_id, data.title, data.content, data.sign)
)
return new_id
except IntegrityError:
retries += 1
raise RuntimeError("Failed to produce an ID. Pool is depleted or near depletion.")
class Timestamps(BaseModel):
created_at: PastDatetime
updated_at: PastDatetime
async def lookup_paste(id: str) -> Paste | None:
data = await database.fetch_one(
"""
SELECT *
FROM pastes
WHERE id = ?
""",
(id,)
)
if not data:
return None
return Paste(**data)
class NewPaste(PasteBase, PasteContent):
pass
async def lookup_paste_raw(id: str) -> str | None:
data = await database.fetch_one(
"""
SELECT content
FROM pastes
WHERE id = ?
""",
(id,)
)
if not data:
return None
return data["content"]
class PastePreview(PasteID, PasteBase, Timestamps):
pass
async def modify_paste(id: str, new_data: NewPaste) -> bool:
print(vars(new_data))
i = await database.execute(
"""
UPDATE pastes
SET title = ?, content = ?, sign = ?, updated_at = datetime('now')
WHERE id = ?
""",
(new_data.title, new_data.content, new_data.sign, id)
)
return bool(i)
class Paste(PastePreview, PasteContent):
pass
class Page(BaseModel, Generic[T]):
items: list[T]
total: int
page: int
page_size: int
async def remove_paste(id: str) -> bool:
i = await database.execute(
"""
DELETE
FROM pastes
WHERE id = ?
""",
(id,)
)
return bool(i)
Поддерживает Markdown
0% или .
You are about to add 0 people to the discussion. Proceed with caution.
Сначала завершите редактирование этого сообщения!
Пожалуйста, зарегистрируйтесь или чтобы прокомментировать