هدر، کوکی، فرم، فایل و CORS

۱. هدرها و middleware

@app.middleware('http')
async def add_timing_and_request_id(
    request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
    started = time.perf_counter()
    request_id = request.headers.get('x-request-id') or uuid.uuid4().hex[:12]
    response = await call_next(request)
    response.headers['X-Request-ID'] = request_id
    response.headers['X-Process-Time'] = f'{(time.perf_counter() - started) * 1000:.1f}ms'
    return response


GREETINGS = {'fa': 'Salam', 'en': 'Hello', 'de': 'Hallo'}


@app.get('/hello')
def hello(accept_language: Annotated[str | None, Header()] = None) -> dict[str, str]:
    language = (accept_language or 'en')[:2]
    return {'greeting': GREETINGS.get(language, GREETINGS['en']), 'language': language}


@app.get('/client')
def client_info(
    user_agent: Annotated[str | None, Header()] = None,
    x_forwarded_for: Annotated[str | None, Header()] = None,
) -> dict[str, str | None]:
    return {'user_agent': user_agent, 'x_forwarded_for': x_forwarded_for}


@app.get('/courses')
def courses(response: Response) -> list[str]:
    response.headers['Cache-Control'] = 'public, max-age=60'  # browsers may keep it 60 s
    return ['Python', 'FastAPI']

خواندن هدر و یک middleware برای همه پاسخ‌ها: نمونه 01_headers/app.py

• هدرهای درخواست با Header() خوانده می‌شوند؛ accept_language یعنی هدر Accept-Language.
• هدر پاسخ یک endpoint با پارامتر response: Response تنظیم می‌شود.
• برای همه پاسخ‌ها یک middleware می‌نویسیم که دور هر درخواست اجرا می‌شود. هدرهای X-Request-ID (دنبال کردن یک درخواست در لاگ‌ها) و X-Process-Time نمونه‌های رایج‌اند.

۲. کوکی‌ها

@app.get('/courses/{course_id}')
def view_course(
    course_id: int, response: Response, recent: Annotated[str | None, Cookie()] = None
) -> dict:
    ids = [int(x) for x in recent.split('-') if x.isdigit()] if recent else []
    ids = [course_id] + [x for x in ids if x != course_id]
    response.set_cookie(
        'recent', '-'.join(map(str, ids[:3])), max_age=7 * 24 * 3600, httponly=True, samesite='lax'
    )
    return {'course': COURSES.get(course_id, '?')}


@app.get('/recently-viewed')
def recently_viewed(recent: Annotated[str | None, Cookie()] = None) -> list[str]:
    if not recent:
        return []
    return [COURSES.get(int(x), '?') for x in recent.split('-') if x.isdigit()]


@app.delete('/recently-viewed')
def forget(response: Response) -> dict[str, str]:
    response.delete_cookie('recent')
    return {'status': 'forgotten'}

«دوره‌های اخیراً دیده‌شده» در کوکی: نمونه 02_cookies/app.py

مرورگرها کوکی را خودکار می‌فرستند؛ راحت است اما مراقبت در برابر CSRF لازم دارد. برنامه‌های موبایل و اسکریپت‌ها معمولاً توکن Bearer می‌فرستند.

۳. داده فرم

فرم HTML به جای JSON این را می‌فرستد: name=Sara&email=.... فیلدها را تکی با Form() یا یک مدل کامل Pydantic می‌خوانیم:

class ContactForm(BaseModel):
    name: str = Field(min_length=2)
    email: EmailStr
    message: str = Field(min_length=10, max_length=1000)
    newsletter: bool = False


@app.get('/', response_class=HTMLResponse)
def page() -> str:
    return """
    <form method="post" action="/contact">
      <input name="name" placeholder="name"><br>
      <input name="email" placeholder="email"><br>
      <textarea name="message" placeholder="at least 10 characters"></textarea><br>
      <label><input type="checkbox" name="newsletter" value="true"> newsletter</label><br>
      <button>Send</button>
    </form>"""


@app.post('/contact', status_code=201)
def contact(data: Annotated[ContactForm, Form()]) -> dict:
    MESSAGES.append(data.model_dump())
    return {'received': data.model_dump(), 'total_messages': len(MESSAGES)}

فرم HTML و یک مدل به عنوان فرم: نمونه 03_form_data/app.py

۴. بارگذاری فایل با UploadFile

@app.post('/upload')
async def upload(file: UploadFile, title: Annotated[str, Form()] = 'untitled') -> dict:
    target = UPLOAD_DIR / Path(file.filename or 'file').name  # .name drops any folders
    size = 0
    with target.open('wb') as out:
        while chunk := await file.read(64 * 1024):  # 64 KB at a time
            size += len(chunk)
            out.write(chunk)
    return {
        'title': title,
        'filename': file.filename,
        'content_type': file.content_type,
        'bytes': size,
        'saved_as': str(target),
    }


@app.post('/upload-many')
async def upload_many(
    files: Annotated[list[UploadFile], File(description='several files')],
) -> list[dict]:
    return [{'filename': f.filename, 'size': f.size, 'type': f.content_type} for f in files]

یک فایل، چند فایل، خواندن تکه‌به‌تکه: نمونه 04_file_upload/app.py

UploadFile نام فایل، نوع محتوا، اندازه و متدهای ناهمگام read و seek را دارد. Starlette فایل‌های بزرگ را روی دیسک نگه می‌دارد، پس خواندن تکه‌به‌تکه حافظه را کم مصرف می‌کند. Path(file.filename).name پوشه‌ها را از نام حذف می‌کند و جلوی نامی مثل ../../etc/passwd را می‌گیرد.

۵. بررسی فایل: به هیچ چیز اعتماد نکنید

UPLOAD_DIR = Path('uploads')
UPLOAD_DIR.mkdir(exist_ok=True)
MAX_BYTES = 1 * 1024 * 1024  # 1 MB for the demo
ALLOWED = {
    'application/pdf': ('.pdf', b'%PDF-'),
    'image/png': ('.png', b'\x89PNG\r\n\x1a\n'),
    'image/jpeg': ('.jpg', b'\xff\xd8\xff'),
    'text/plain': ('.txt', None),
}
app = FastAPI()


@app.post('/materials', status_code=status.HTTP_201_CREATED)
async def upload_material(file: UploadFile) -> dict:
    if file.content_type not in ALLOWED:
        raise HTTPException(
            status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, detail=f'{file.content_type} not allowed'
        )
    extension, magic = ALLOWED[file.content_type]
    if magic is not None:
        head = await file.read(len(magic))
        await file.seek(0)
        if head != magic:
            raise HTTPException(
                status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, detail='Content does not match the type'
            )

    stored = UPLOAD_DIR / f'{uuid.uuid4().hex}{extension}'
    size = 0
    try:
        with stored.open('wb') as out:
            while chunk := await file.read(64 * 1024):
                size += len(chunk)
                if size > MAX_BYTES:
                    raise HTTPException(
                        status.HTTP_413_CONTENT_TOO_LARGE, detail='File is larger than 1 MB'
                    )
                out.write(chunk)
    except BaseException:
        stored.unlink(missing_ok=True)  # no half-written files left behind
        raise
    return {'original_name': file.filename, 'stored_as': stored.name, 'bytes': size}

نوع، بایت‌های جادویی و اندازه: نمونه 05_file_validation/app.py

1. content_type را کلاینت اعلام می‌کند و ممکن است دروغ بگوید؛ فهرست سفید نوع‌ها را بررسی کنید (۴۱۵).
2. پس اولین بایت‌ها را هم نگاه کنید: PDF با %PDF- شروع می‌شود، PNG با \x89PNG و JPEG با \xff\xd8\xff.
3. اندازه را هنگام خواندن محدود کنید و زود متوقف شوید (۴۱۳).
4. فایل را با نام ساخته‌شده (uuid4) ذخیره کنید، هرگز با نام کلاینت.
5. اگر وسط کار چیزی شکست خورد، فایل نیمه‌کاره را پاک کنید.

05 ok.pdf   201 558b239b10d94afa91d54a551d92dff5.pdf
05 fake.pdf 415 Content does not match the type
05 v.exe    415 application/x-msdownload not allowed
05 big.txt  413 File is larger than 1 MB

۶. فایل ایستا یا دانلود خصوصی؟

app.mount('/static', StaticFiles(directory='static')) یک پوشه کامل را همان‌طور که هست ارائه می‌کند: تصویر، CSS و PDFهای عمومی. اما عمومی است و احراز هویت ندارد. فایل‌های خصوصی، مثل جزوه‌هایی که فقط دانشجویان ثبت‌نام‌شده باید بخوانند، باید از یک endpoint بگذرند که مجوز را بررسی می‌کند و بعد یک FileResponse برمی‌گرداند. filename= هدر Content-Disposition را اضافه می‌کند تا مرورگر فایل را با نامی خوب دانلود کند.

۷. CORS: آیا صفحه‌ای از مبدأ دیگر می‌تواند این API را صدا بزند؟

«مبدأ» یعنی طرح، میزبان و درگاه. http://localhost:5500 و http://127.0.0.1:8000 دو مبدأ متفاوت‌اند. مرورگر اجازه نمی‌دهد جاوااسکریپت پاسخ مبدأ دیگری را بخواند، مگر اینکه API برای آن مبدأ هدر Access-Control-Allow-Origin بفرستد. برای بعضی درخواست‌ها مرورگر اول یک درخواست OPTIONS «پیش‌پرواز» می‌فرستد.

ALLOWED_ORIGINS = ['http://localhost:5500', 'http://127.0.0.1:5500']

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,  # never ['*'] together with allow_credentials=True
    allow_credentials=True,
    allow_methods=['GET', 'POST', 'PATCH', 'DELETE'],
    allow_headers=['Authorization', 'Content-Type'],
)


@app.get('/courses')
def courses() -> list[dict[str, str | int]]:
    return [{'id': 1, 'title': 'Python'}, {'id': 2, 'title': 'FastAPI'}]

CORSMiddleware: نمونه 07_cors/app.py

۸. انواع پاسخ

@app.get('/courses', response_model_exclude_none=True)
def list_courses() -> list[Course]:
    return [Course(**c) for c in COURSES]  # the null `note` disappears from the JSON


@app.get('/old-courses')
def old_address() -> RedirectResponse:
    return RedirectResponse('/courses', status_code=status.HTTP_308_PERMANENT_REDIRECT)


@app.get('/ping', response_class=PlainTextResponse)
def ping() -> str:
    return 'pong'


@app.post('/reports', status_code=status.HTTP_202_ACCEPTED)
def start_report() -> JSONResponse:
    return JSONResponse({'status': 'queued'}, status_code=202, headers={'Location': '/reports/42'})


def csv_rows() -> Iterator[str]:
    buffer = io.StringIO()
    writer = csv.writer(buffer)
    writer.writerow(['id', 'title', 'price'])
    for course in COURSES:
        writer.writerow([course['id'], course['title'], course['price']])
        yield buffer.getvalue()
        buffer.seek(0)
        buffer.truncate()
    yield buffer.getvalue()


@app.get('/courses.csv')
def export_csv() -> StreamingResponse:
    return StreamingResponse(
        csv_rows(),
        media_type='text/csv',
        headers={'Content-Disposition': 'attachment; filename="courses.csv"'},
    )

تغییر مسیر، متن ساده، ۲۰۲ و CSV جریانی: نمونه 08_responses/app.py

خروجی CSV از یک مولد (فصل اول) جریان پیدا می‌کند؛ حتی یک جدول بسیار بزرگ لازم نیست کامل در حافظه جا شود.

۹. پروژه: فایل‌های درسی

مدل تازه CourseMaterial با نسخه مهاجرت ۰۰۰۴ اضافه می‌شود. بایت‌های فایل روی دیسک‌اند و فقط اطلاعات آنها در پایگاه داده:

class LocalStorage:
    def __init__(self, root: Path, max_bytes: int) -> None:
        self.root = root
        self.max_bytes = max_bytes

    def path_for(self, stored_name: str) -> Path:
        return self.root / stored_name

    def check_type(self, upload: UploadFile) -> str:
        """Validate the declared type AND the real first bytes; return the extension."""
        content_type = upload.content_type or ''
        if content_type not in ALLOWED_TYPES:
            allowed = ', '.join(ALLOWED_TYPES)
            raise FileValidationError(f'{content_type!r} is not allowed. Allowed: {allowed}')
        extension, magic = ALLOWED_TYPES[content_type]
        if magic is not None:
            head = upload.file.read(len(magic))
            upload.file.seek(0)
            if head != magic:
                raise FileValidationError(f'The file content is not really {content_type}')
        return extension

    def save(self, upload: UploadFile, extension: str) -> tuple[str, int]:
        """Copy the upload in chunks under a random name. Returns (name, size)."""
        self.root.mkdir(parents=True, exist_ok=True)
        stored_name = f'{uuid.uuid4().hex}{extension}'
        target = self.path_for(stored_name)
        size = 0
        try:
            with target.open('wb') as out:
                while chunk := upload.file.read(CHUNK_SIZE):
                    size += len(chunk)
                    if size > self.max_bytes:
                        raise FileTooLargeError(f'The file is larger than {self.max_bytes} bytes')
                    out.write(chunk)
        except BaseException:
            target.unlink(missing_ok=True)
            raise
        return stored_name, size

    def delete(self, stored_name: str) -> None:
        self.path_for(stored_name).unlink(missing_ok=True)

ذخیره امن روی دیسک: 09_project/app/storage.py

    def _ensure_can_read(self, course: Course, user: User) -> None:
        """Staff of the course, or a student who is (or was) enrolled in it."""
        if can_manage_course(course, user):
            return
        enrollment = self.db.scalar(
            select(Enrollment).where(
                Enrollment.course_id == course.id, Enrollment.student_id == user.id
            )
        )
        if enrollment is None or enrollment.status == 'cancelled':
            raise PermissionDeniedError('Only students of this course can see its materials')

چه کسی می‌تواند فایل‌ها را ببیند: 09_project/app/services.py

در کنار این‌ها CORS از روی تنظیم CORS_ORIGINS فعال می‌شود و هر پاسخ هدرهای X-Request-ID و X-Process-Time دارد. درخواست‌های بارگذاری Bruno از فایل‌های نمونه پوشه bruno/files/ استفاده می‌کنند؛ یکی از آنها یک PDF جعلی است که باید ۴۱۵ بگیرد.

۱۰. خلاصه فصل