paper-core 0.1.0

Core runtime library for the PaperDraft framework

paper-core

Runtime library for the PaperDraft framework. Provides auth, database, encryption, email, middleware, error handling, and audit logging — all importable under the paper.core namespace.

Status: pre-release · v0.1 in progress · Paper Plane Consulting LLC

---

Table of Contents

1. Installation
2. Local Development
3. Running Tests
4. Module Reference
5. Extending Core

---

Installation

pip install paper-core

Requirements: Python 3.11+

---

Local Development

git clone <repo-url>
cd paper-core

python -m venv .venv
source .venv/bin/activate     # macOS/Linux
.venv\Scripts\activate        # Windows

pip install -e ".[dev]"

The -e . (editable install) is required — it registers the paper namespace package so that from paper.core.X import Y resolves correctly.

---

Running Tests

pytest

Run a specific module:

pytest tests/test_auth.py -v

Test suite coverage

File	Covers
test_errors.py	ErrorHandler, ErrorMessage
test_security.py	Hasher, Crypto, RSACrypto, Pem, Encoding
test_audit.py	Audit, AuditAction, AuditOutcome
test_auth.py	Credentials, Signature, Password, Auth, Key, Claims, Authenticate, Authorize, LoginAttemptLimit
test_middleware.py	HipaaResponseHeaders, RequestIdMiddleware, RequestLoggingMiddleware
test_db.py	FilterType, ConnectionPoolConfig, _relationship_fields, MultiTenantPoolManager
test_email.py	EmailTheme, Info, Body, Message, Subject, MimeType

Note: Postgres CRUD tests (create, retrieve, update, delete) require a live PostgreSQL connection and are not included. Run those as integration tests against a test database.

---

Module Reference

paper.core.auth

JWT signing, password hashing, and FastAPI authentication dependencies.

from paper.core.auth import (
    Auth,                   # signs JWT access + refresh token pairs
    Password,               # Argon2 hash and verify
    Authenticate,           # FastAPI dep — validates JWT from header or cookie
    Authorize,              # FastAPI dep — validates JWT + enforces RBAC roles
    LoginAttemptLimit,      # FastAPI dep — rate limits login attempts per IP
    set_auth_params,        # call once at startup
    set_login_attempt_params,
    Claims, Key,            # token decode utilities
    Credentials, Signature, # request/response models
    Algorithm, TokenType, ClaimsKey, AuthErrorMessage,
)

Startup wiring (in dependencies.py):

from paper.core.auth import set_auth_params, set_login_attempt_params
from paper.core.auth.enums import Algorithm
from datetime import timedelta

set_auth_params({
    "public_key":     config.ENCRYPTION.PUBLIC_KEY,
    "excluded_paths": ["/auth/login", "/auth/refresh", "/health"],
    "alg":            Algorithm.RS256,
})

set_login_attempt_params(max_attempts=5, lockout_duration=timedelta(minutes=15))

Route usage:

from paper.core.auth import Authenticate, Authorize

@router.get("/me")
async def me(claims = Depends(Authenticate())):
    return claims

@router.delete("/{id}")
async def delete(claims = Depends(Authorize(["admin"]))):
    ...

---

paper.core.db

Async SQLAlchemy PostgreSQL repository with optional multi-tenant pool management.

from paper.core.db import (
    Postgres,                   # async SQLAlchemy repository
    ConnectionPoolConfig,       # pool settings
    Repository,                 # abstract base — extend to add engines
    FilterType,                 # EQUAL, LIKE, IN, NOT_IN, etc.
    MultiTenantPoolManager,     # one pool per tenant DSN
    MultiTenantDbDependency,    # FastAPI dep — resolves tenant DB from JWT
)

Single-tenant setup:

from paper.core.db import Postgres, ConnectionPoolConfig

db = Postgres(
    connection_string = config.POSTGRES_CONN_STRING,
    config            = ConnectionPoolConfig(
        future=True, size=10, max_overflow=5,
        recycle_after=3600, timeout=30, pre_ping=True,
    ),
)

Filtering:

results = await db.retrieve(
    UserEntity, UserModel,
    filter={FilterType.EQUAL: {"is_active": True}},
)

---

paper.core.security

RSA-OAEP encryption and random code generation.

from paper.core.security import Crypto, RSACrypto, Hasher, Pem, Encoding

# Static utility (key passed per call)
cipher = Crypto.encrypt_urlsafe(dsn, public_key_b64)
dsn    = Crypto.decrypt_urlsafe(cipher, private_key_b64)

# Instance-based (keys injected once)
crypto = RSACrypto(public_key_b64, private_key_b64)
cipher = crypto.encrypt_urlsafe(dsn)

# Random codes (invites, resets)
code = Hasher.generate(8)   # e.g. "aB3xZ9mK"

Encoding PEM keys for .env storage:

openssl genrsa -out private.pem 2048
openssl rsa -in private.pem -pubout -out public.pem
base64 -w 0 private.pem   # → ENCRYPTION_PRIVATE_KEY
base64 -w 0 public.pem    # → ENCRYPTION_PUBLIC_KEY

Or with Pem:

from paper.core.security import Pem
print(Pem.to_base64("private.pem"))

---

paper.core.middleware

from paper.core.middleware import (
    HipaaResponseHeaders,       # X-Frame-Options, CSP, HSTS, etc.
    RequestIdMiddleware,        # X-Request-ID on every request
    RequestLoggingMiddleware,   # structured request/response logging
)

Registration order in main.py:

app.add_middleware(CORSMiddleware, ...)
app.add_middleware(HipaaResponseHeaders)
app.add_middleware(RequestLoggingMiddleware)
app.add_middleware(RequestIdMiddleware)    # executes first

Access the request ID downstream:

request.state.request_id

---

paper.core.errors

from paper.core.errors import ErrorHandler, ErrorMessage

ErrorHandler.handle(404, f"{ErrorMessage.NOT_FOUND.value} user {id}")
ErrorHandler.handle(401, "Unauthorized", {"WWW-Authenticate": "Bearer"})

All service layers raise through ErrorHandler — never construct HTTPException directly.

---

paper.core.audit

Generic audit logger. Accepts any DB, entity, and model — no framework-level schema dependency.

from paper.core.audit import Audit, AuditAction, AuditOutcome

audit = Audit(db=db, entity=AuditLogEntity, model=AuditLogModel)

await audit.log(
    event_type    = LoginEvent.SUCCESS,
    action        = AuditAction.ATTEMPT,
    outcome       = AuditOutcome.SUCCESS,
    email         = credentials.email,
    ip_address    = request.client.host,
    user_agent    = request.headers.get("user-agent"),
)

Audit failures are swallowed silently and logged — they never block the calling operation.

---

paper.core.email

SMTP email with framework lifecycle templates and themeable HTML.

from paper.core.email import (
    Server, Info, Body, Message,
    Subject, EmailTheme, EmailBodyParam,
)

server = Server(host, port, username, password)
sender = Info("My App", "[e-mail]")

body = Body(
    subject = Subject.RESET_PASSWORD,
    data    = {
        EmailBodyParam.REDIRECT_URL.value: reset_url,
        EmailBodyParam.RESET_CODE.value:   code,
    },
)

msg = Message(
    sender    = sender,
    recipient = Info(user.name, user.email),
    subject   = Subject.RESET_PASSWORD.value,
    text      = body.text,
    html      = body.html,
)

server.send(msg)

Theming via environment variables:

EMAIL_THEME_PRIMARY_COLOR=#0057a8
EMAIL_THEME_BUTTON_COLOR=#0057a8
EMAIL_THEME_LOGO_URL=https://cdn.myapp.com/logo.png
EMAIL_THEME_COMPANY_NAME=My App

---

Extending Core

All major components are built on abstract base classes — extend them to swap implementations without changing the calling code.

Custom Database Engine

from paper.core.db.base import Repository, FilterType

class MySQLRepository(Repository[T, M]):
    async def create(self, entity, model, data): ...
    async def retrieve(self, entity, model, filter=None): ...
    async def single(self, entity, model, id): ...
    async def update(self, entity, model, id, data): ...
    async def delete(self, entity, id): ...

Custom Encryption Algorithm

from paper.core.security.base import BaseCrypto

class AESCrypto(BaseCrypto):
    def encrypt(self, value: str) -> str: ...
    def decrypt(self, cipher: str) -> str: ...
    def encrypt_urlsafe(self, value: str) -> str: ...
    def decrypt_urlsafe(self, cipher: str) -> str: ...
    def encrypt_raw(self, value: str) -> bytes: ...
    def decrypt_raw(self, cipher_bytes: bytes) -> str: ...

Custom Email Provider

from paper.core.email.base import BaseEmailService

class SendGridEmailService(BaseEmailService):
    def send(self, subject, recipient_name, recipient_email, data) -> bool: ...