Pular para o conteúdo principal

Blog

Desenvolvendo uma Aplicação em FastAPI para Compartilhamento de Arquivos - Parte 1

André Felipe Dias

Comentários

Também disponível em:

English

Desenvolvendo uma Aplicação em FastAPI para Compartilhamento de Arquivos - Parte 1

Este é um tutorial de construção de uma aplicação WEB para compartilhamento de arquivos, bastante útil nos casos em que não há acesso via SSH. A aplicação será desenvolvida a partir do gabarito do projeto FastAPI mínimo, o que economizará bastante tempo e esforço pois partiremos de uma estrutura básica de projeto já pronta.

Introdução

Imagine a seguinte situação: você precisa transferir um arquivo para ou de um servidor mas não tem acesso via SSH e, portanto, não pode usar scp ou rsync. Configurar o SSH pode até ser possível, mas é muito trabalhoso para uma transferência que você precisa fazer apenas uma vez.

Uma solução é usar algum tipo de armazenamento em nuvem para fazer o upload e o download do arquivo através de uma REST API. Mas ao invés de usar DropBox, Google Drive ou algo parecido, vamos desenvolver nossa própria aplicação em FastAPI.

Especificações

A aplicação deve ter as seguintes características:

  • A resposta ao upload bem sucedido de um arquivo é a URL para download desse mesmo arquivo.
  • Cada arquivo tem uma URL de download único, independente do nome ou conteúdo do arquivo.
  • O upload e o download são anônimos. Não é necessário fazer autenticação.
  • O tamanho máximo de um upload é de 5 MiB.
  • Os arquivos são apagados automaticamente após 60 minutos.
  • O número máximo de uploads por IP é de 5 por hora.
  • Não há limite para número de downloads.
  • O arquivo pode ser deletado manualmente através de um DELETE para a URL de download.

Detalhamento Técnico

Endpoints

Vamos precisar de três endpoints: um para upload, um para download e um para delete. Para o endpoint de download, vamos criar um caminho no formato /{token}/{filename}, em que token é uma string aleatória e filename é o nome do arquivo original:

Método Caminho Descrição
POST / Upload de um arquivo
GET /{token}/{filename} Download de um arquivo
DELETE /{token}/{filename} Remove um arquivo

Considerando https://transfer.pronus.xyz como endereço da aplicação, o upload de um arquivo /tmp/numbers_123.txt usando HTTPie seria feito assim:

$ http --form POST https://transfer.pronus.xyz file@/tmp/numbers_123.txt
HTTP/1.1 201 Created
...
Location: https://transfer.pronus.xyz/kCLJQz5vxmE/numbers_123.txt
...

https://transfer.pronus.xyz/kCLJQz5vxmE/numbers_123.txt

O download do arquivo seria dessa forma:

$ http https://transfer.pronus.xyz/kCLJQz5vxmE/numbers_123.txt
HTTP/1.1 200 OK
Content-Type: text/plain; charset=utf-8
...

E a remoção do arquivo seria pelo comando:

$ http DELETE https://transfer.pronus.xyz/kCLJQz5vxmE/numbers_123.txt
HTTP/1.1 204 No Content
...

Refazendo os mesmos comandos com curl, teríamos:

$ curl --form file=@/tmp/numbers_123.txt https://transfer.pronus.xyz/

$ curl https://transfer.pronus.xyz/kCLJQz5vxmE/numbers_123.txt

$ curl -X DELETE https://transfer.pronus.xyz/kCLJQz5vxmE/numbers_123.txt

Armazenamento

Em uma aplicação mais séria, o armazenamento dos arquivos seria feito em um serviço de armazenamento em nuvem como S3 ou Google Cloud Storage. Mas para simplificar o projeto, vamos usar o próprio sistema de arquivos do servidor.

O diretório de armazenamento será configurado através de uma variável de ambiente chamada UPLOAD_DIR. Dentro desse diretório, criaremos subdiretórios com os nomes do tokens. Cada um desses subdiretórios conterá apenas o arquivo enviado, com o mesmo nome do arquivo original.

Por exemplo, ser for feito o upload do arquivo numbers_123.txt, o token aleatório for kCLJQz5vxmE e o diretório de armazenamento for /dev/shm/transfer, o arquivo será armazenado em /dev/shm/transfer/kCLJQz5vxmE/numbers_123.txt.

Agendamento da Remoção

Logo que um arquivo é recebido e armazenado, um agendamento é criado para que o arquivo seja removido após 60 minutos. Se o arquivo for removido manualmente através de um DELETE /{token}/{filename}, o agendamento deve ser cancelado.

Vamos usar o APScheduler para controlar os agendamentos por ser uma solução simples e que não requer nenhum outro serviço ou configuração adicionais no modo de armazenamento em memória.

Como os agendamentos serão mantidos apenas em memória, tanto os arquivos quando os agendamentos serão perdidos em caso de falha ou reinício do servidor. Isso é aceitável para o nosso caso porque a finalidade da aplicação é apenas fornecer uma solução transitória para transferência de arquivos e não armazenar arquivos de forma permanente.

Verificação do Tamanho Máximo do Arquivo

O tamanho do arquivo sendo baixado é declarado no cabeçalho Content-Length. Porém, essa informação não é confiável pois pode ser forjada facilmente. Será necessário uma outra verificação adicional para garantir que o arquivo não exceda o tamanho máximo.

Limite de Uploads por IP

Vamos limitar o número de uploads para evitar abusos. Como o acesso é anônimo, a aplicação não possuirá nenhum banco de dados de usuário e senha. A identificação será feita por IP, com um limite de 5 uploads por hora por IP.

Apesar de ser possível fazer essa verificação pela aplicação, é melhor deixar essa tarefa para o servidor web por ser mais eficiente e mais simples de implementar.

Instanciação do Gabarito de Projeto

Vamos começar instanciando o gabarito do projeto FastAPI mínimo. A melhor maneira é usando pipx:

$ pipx run cookiecutter gh:andredias/perfect_python_project \
  -c fastapi-minimum
[1/8] author (): André Felipe Dias
[2/8] email (): andre.dias@email.com
[3/8] project_name (Perfect Python Project): Transfer
[4/8] project_slug (transfer):
[5/8] python_version (3.11):
[6/8] line_length (79): 100
[7/8] Select version_control
    1 - hg
    2 - git
    Choose from [1/2] (1):
[8/8] github_respository_url ():
Creating virtualenv transfer in /home/andre/projetos/tutoriais/transfer/.venv
Using virtualenv: /home/andre/projetos/tutoriais/transfer/.venv
Updating dependencies
...

Package operations: 66 installs, 0 updates, 0 removals

• Installing certifi (2023.7.22)
• Installing charset-normalizer (3.2.0)
...

Writing lock file
ruff --silent --exit-zero --fix .
blue .
reformatted transfer/main.py
reformatted transfer/config.py

All done! ✨ 🍰 ✨
2 files reformatted, 8 files left unchanged.
adding .dockerignore
adding .github/workflows/continuous_integration.yml
adding .hgignore
...

Em seguida, vamos instalar algumas dependências adicionais: aiofiles, apscheduler e python-multipart como dependências de projeto, e types-aiofiles para linting. O último é uma dependência de desenvolvimento e fica em um grupo separado:

$ poetry add  aiofiles=="*" apscheduler=="*" python-multipart=="*"
$ poetry add --group dev types-aiofiles=="*"

A estrutura de diretórios do projeto transfer fica assim:

transfer
├── Caddyfile
├── docker-compose.dev.yml
├── docker-compose.yml
├── Dockerfile
├── hypercorn.toml
├── Makefile
├── poetry.lock
├── pyproject.toml
├── README.rst
├── sample.env
├── scripts
│   └── install_hooks.sh
├── tests
│   ├── conftest.py
│   ├── __init__.py
│   ├── routers
│   │   ├── __init__.py
│   │   └── test_hello.py
│   └── test_logging.py
└── transfer
    ├── config.py
    ├── exception_handlers.py
    ├── __init__.py
    ├── logging.py
    ├── main.py
    ├── middleware.py
    ├── resources.py
    └── routers
        ├── hello.py
        └── __init__.py

Configuração

Além das variáveis de configuração herdadas do projeto mínimo em FastAPI, vamos precisar incluir as seguintes variáveis:

  • BUFFER_SIZE: Tamanho do buffer usado na verificação manual do tamanho do arquivo. O tamanho padrão será de 1 MiB.
  • FILE_SIZE_LIMIT: Tamanho máximo do arquivo. O tamanho padrão será de 5 MiB.
  • UPLOAD_DIR: Diretório de armazenamento dos arquivos. O diretório padrão será /tmp/transfer_files.
  • TOKEN_LENGTH: Tamanho do token usado na URL de download. O tamanho padrão será de 8 bytes.
  • TIMEOUT_INTERVAL: Intervalo de tempo em segundos para remoção automática dos arquivos. O intervalo padrão será de 3.600 segundos (1 hora).

O arquivo de configuração fica:

import os
from pathlib import Path
from typing import Final
from dotenv import load_dotenv
load_dotenv()
ENV: str = os.getenv('ENV', 'production').lower()
if ENV not in ('production', 'development', 'testing'):
    raise ValueError(
        f'ENV={ENV} is not valid. ' "It should be 'production', 'development' or 'testing'"
    )
DEBUG: bool = ENV != 'production'
TESTING: bool = ENV == 'testing'
os.environ['LOGURU_LEVEL'] = os.getenv('LOG_LEVEL') or (DEBUG and 'DEBUG') or 'INFO'
os.environ['LOGURU_DEBUG_COLOR'] = '<fg #777>'
REQUEST_ID_LENGTH = int(os.getenv('REQUEST_ID_LENGTH', '8'))
PYGMENTS_STYLE = os.getenv('PYGMENTS_STYLE', 'github-dark')
MiB: Final[int] = 2**20
BUFFER_SIZE: int = int(os.getenv('BUFFER_SIZE') or 1 * MiB)
FILE_SIZE_LIMIT: int = int(os.getenv('FILE_SIZE_LIMIT') or 5 * MiB)
UPLOAD_DIR: Path = Path(os.getenv('UPLOAD_DIR', '/tmp/transfer_files'))  # noqa: S108
TOKEN_LENGTH: int = int(os.getenv('TOKEN_LENGTH') or 8)
TIMEOUT_INTERVAL: int = int(os.getenv('TIMEOUT_INTERVAL_MIN') or 3_600)  # in seconds

Manipulação de Arquivos Baixados

Tendo em mente que teremos ações para POST, GET e DELETE de arquivos, também precisaremos de funções para gravar, remover e verificar a existência de arquivos baixados. Implementar essas funcionalidades diretamente no endpoint prejudicaria a testabilidade e a reusabilidade das funções. Por isso, vamos agrupá-las em um módulo separado chamado file_utils.py.

Verificação da Existência de Um Arquivo Baixado

Para verificar a existência de um arquivo, temos:

from . import config


def file_exists(token: str, filename: str) -> bool:
    return (config.UPLOAD_DIR / token / filename).exists()

Remoção de Arquivos

Para remover um arquivo:

from shutil import rmtree

from . import config


def remove_file(token: str) -> None:  # noqa: ARG001
    """
    Remove a file and its parent directory
    """
    rmtree(config.UPLOAD_DIR / token)

O comentário # noqa: ARG001 é uma diretiva para o linter ignorar que filename é declarado, mas não é utilizado. O nome do arquivo não é necessário porque o nome do diretório é único, definido pelo token. Como só há um arquivo por diretório e o arquivo será removido, então basta remover o diretório com o arquivo dentro. No entanto, o nome do arquivo será mantido como parâmetro para manter a consistência com as outras funções.

Remoção de Arquivos de Tempo de Armazenamento Expirado

Uma outra função que será útil futuramente é a que remove todos os arquivos que já passaram do tempo limite. Essa função será chamada pelo APScheduler para garantir que o armazenamento será limpo periodicamente mesmo que haja alguma falha no servidor.

from time import time

from . import config


def remove_expired_files() -> None:
    """
    Remove files that have been stored for too long
    """
    timeout_ref = time() - config.TIMEOUT_INTERVAL
    for path in config.UPLOAD_DIR.glob('*/*'):
        if path.stat().st_mtime < timeout_ref:
            remove_file(*path.relative_to(config.UPLOAD_DIR).parts)

Armazenamento de Arquivos Baixados

E por fim, a função mais importante, a que salva o arquivo. Ela é mais complexa porque durante o processo de salvamento é necessário verificar o tamanho real do arquivo, já que o tamanho declarado no cabeçalho Content-Length pode ser forjado.

import secrets
from pathlib import Path
from typing import Protocol
import aiofiles
from . import config
class Readable(Protocol):
    async def read(self, size: int) -> bytes:
        ...
async def save_file(file: Readable) -> tuple[str, str]:
    """
    save the file
    Content-Length header is not reliable to prevent overflow
    see: https://github.com/tiangolo/fastapi/issues/362#issuecomment-584104025
    """
    # UploadFile has filename and aiofile has name attribute
    filename = Path(getattr(file, 'filename', '') or getattr(file, 'name', 'no-name')).name
    token = secrets.token_urlsafe(config.TOKEN_LENGTH)
    path = config.UPLOAD_DIR / token / filename
    path.parent.mkdir(parents=True)
    real_file_size = 0
    overflow = False
    async with aiofiles.open(path, 'wb') as out_file:
        while content := await file.read(config.BUFFER_SIZE):
            real_file_size += len(content)
            if overflow := real_file_size > config.FILE_SIZE_LIMIT:
                break
            await out_file.write(content)
    if overflow:
        remove_file(token, filename)
        raise OSError(f'File {filename} is too large')
    return token, filename

A classe Readable (linha 10) foi criada para definir o tipo do parâmetro file com uma interface mínima que servirá tanto para o UploadFile do FastAPI quanto para um aiofile, usado nos testes.

O nome do arquivo (filename) será inferido a partir de file (linha 22). Contudo, UploadFile possui o atributo filename enquanto aiofile possui o atributo name.

É também durante a gravação que o token é gerado, baseado na função token_urlsafe do módulo secrets, que gera uma string aleatória de tamanho configurável que pode ser usada em URLs (linha 23).

A verificação do tamanho real do arquivo é feita enquanto se salva o arquivo por partes. O tamanho de cada parte salva é contabilizada e se o tamanho real do arquivo exceder o limite, o processo de salvamento é interrompido, o arquivo parcialmente salvo é removido e uma exceção é lançada (linhas 29 a 36).

Testes Unitários das Funções de Manipulação de Arquivos

Antes de passar para a próxima parte do projeto, vamos criar testes para as funções de manipulação de arquivos. Vamos usar três testes:

  1. Testar o ciclo de criação, verificação e remoção de um arquivo.
  2. Testar a remoção de arquivos cujos tempos de vida expiraram.
  3. Testar o salvamento de arquivos que excedem o tamanho máximo.

Esses testes estarão no arquivo test_file_utils.py.

Teste de Ciclo de Vida de Arquivos

A ideia aqui é cobrir o ciclo de criação, verificação e remoção do arquivo usando as funções file_exists, remove_file e save_file:

from pathlib import Path
import aiofiles
from transfer.file_utils import file_exists, remove_file, save_file
async def test_file_cycle(tmp_path: Path) -> None:
    assert not file_exists('tralala', 'dummy.txt')
    dummy = tmp_path / 'dummy.txt'
    dummy.write_text('hello world')
    async with aiofiles.open(dummy, 'rb') as file:
        token, filename = await save_file(file)
    assert file_exists(token, filename)
    remove_file(token, filename)
    assert not file_exists(token, filename)

tmp_path é uma fixture do pytest que cria um diretório temporário para os testes. Dessa forma, não precisamos nos preocupar em limpar o diretório de armazenamento ao final de cada teste.

A primeira verificação é de um arquivo não existe (linha 9). Em seguida, criamos um arquivo que será usado para upload (linha 10 e 11). Esse arquivo precisa ser passado através de um parâmetro que permita leitura assíncrona em modo binário, que é a forma como UploadFile (FastAPI) trabalha. Por isso estamos usando o aiofiles (linha 12).

Depois que o arquivo é salvo, verificamos se ele existe (linha 14). Por fim, removemos o arquivo e verificamos se ele não existe mais (linhas 15 e 16).

Teste de Remoção de Arquivos Expirados

import aiofiles
from transfer import config
from transfer.file_utils import file_exists, remove_expired_files, save_file
async def test_remove_expired_files(tmp_path: Path) -> None:
    dummy = tmp_path / 'dummy.txt'
    dummy.write_text('dummy')
    async with aiofiles.open(dummy, 'rb') as file:
        token, filename = await save_file(file)
    assert file_exists(token, filename)
    # no expired files yet
    remove_expired_files()
    assert file_exists(token, filename)
    # fake a time in the  future...
    with patch('transfer.file_utils.time', return_value=time() + config.TIMEOUT_INTERVAL):
        remove_expired_files()
    assert not file_exists(token, filename)

O teste cria e salva um arquivo (linhas 8 a 11). A função que remove os arquivos expirados é chamada (linha 15) mas nada deve mudar porque não se passou tempo suficiente ainda. Em seguida, o tempo é forjado para o futuro (linha 19), a função de remoção é chamada novamente (linha 20) e, dessa vez, o arquivo deve ter sido removido (linha 21).

Teste de Salvamento de Arquivos Grandes

O teste garante que um arquivo que excede o tamanho máximo não é salvo:

from unittest.mock import patch
import aiofiles
from pytest import raises
@patch('transfer.config.FILE_SIZE_LIMIT', 10)
async def test_file_overflow(tmp_path: Path) -> None:
    filename = tmp_path / 'dummy.txt'
    filename.write_text('hello world')
    async with aiofiles.open(filename, 'rb') as file:
        with raises(OSError):
            await save_file(file)

O teste começa forjando o tamanho máximo do arquivo para 10 bytes (linha 7). Em seguida, cria um arquivo de 11 bytes (linha 10) e tenta salvá-lo (linha 13). Como o arquivo excede o tamanho máximo, espera-se que uma exceção do tipo OSError seja lançada (linha 12). Se não for, o teste falha.

Execução dos Testes

Este é o estado da execução dos testes até o momento:

(transfer) $ make test
pytest -x --cov-report term-missing --cov-report html --cov-branch \
    --cov transfer/
========================= test session starts ==========================
platform linux -- Python 3.11.4, pytest-7.4.1, pluggy-1.3.0
rootdir: /home/andre/projetos/tutoriais/transfer
configfile: pyproject.toml
plugins: alt-pytest-asyncio-0.7.2, anyio-4.0.0, cov-4.1.0
collected 8 items

tests/test_file_utils.py ...                                      [ 37%]
tests/test_logging.py ....                                        [ 87%]
tests/routers/test_hello.py .                                     [100%]

---------- coverage: platform linux, python 3.11.4-final-0 -----------
Name                          Stmts  Miss Branch BrPart  Cover   Missing
------------------------------------------------------------------------
transfer/__init__.py              0     0      0      0   100%
transfer/config.py               20     1      2      1    91%   11
transfer/exception_handlers.py   11     0      0      0   100%
transfer/file_utils.py           37     1     12      0    98%   39
transfer/logging.py              25     0      8      2    94%   15->19, 30->32
transfer/main.py                 14     0      2      0   100%
transfer/middleware.py           33     2      4      0    95%   37-38
transfer/resources.py            21     0      4      0   100%
transfer/routers/__init__.py      0     0      0      0   100%
transfer/routers/hello.py         7     0      2      0   100%
------------------------------------------------------------------------
TOTAL                           168     4     34      3    97%
Coverage HTML written to dir htmlcov


========================== 8 passed in 0.51s ===========================

Iniciação do Agendador

A iniciação e término de recursos é feita respectivamente nas funções startup e shutdown no arquivo resources.py. Portanto, vamos alocar o agendador na função startup e cancelar os agendamentos na função shutdown:

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import FastAPI
from loguru import logger
from . import config
from .file_utils import remove_expired_files
from .logging import init_loguru
scheduler = AsyncIOScheduler()
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator:  # noqa: ARG001
    await startup()
    try:
        yield
    finally:
        await shutdown()
async def startup() -> None:
    init_loguru()
    show_config()
    scheduler.start()
    scheduler.add_job(remove_expired_files, 'interval', days=1)
    logger.info('started...')
async def shutdown() -> None:
    scheduler.shutdown()
    logger.info('...shutdown')
def show_config() -> None:
    config_vars = {key: getattr(config, key) for key in sorted(dir(config)) if key.isupper()}
    logger.debug('config vars', **config_vars)

O agendador é instanciado no começo do módulo (linha 12). Na função startup, o agendador é iniciado (linha 27) e um agendamento para a função remove_expired_files é adicionado para executar uma vez por dia (linha 28). Na função shutdown, o agendador é encerrado (linha 33).

Criação dos Endpoints

Agora que já temos as funções de manipulação de arquivos, vamos criar os endpoints no módulo routers/file.py. Antes porém, vamos alterar main.py para incluir o novo roteador:

from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from starlette.middleware.base import BaseHTTPMiddleware
from . import config  # noqa: F401
from .exception_handlers import request_validation_exception_handler
from .middleware import log_request_middleware
from .resources import lifespan
from .routers import file
app = FastAPI(
    title='Transfer',
    lifespan=lifespan,
)
routers = (file.router,)
for router in routers:
    app.include_router(router)
app.add_middleware(BaseHTTPMiddleware, dispatch=log_request_middleware)
app.add_exception_handler(RequestValidationError, request_validation_exception_handler)

As alterações estão nas linhas 9 e 16, com a inclusão do roteador file. O roteador para hello foi removido porque não será mais usado.

GET /{token}/{filename}

Esse é o endpoint mais simples. Basta retornar o arquivo se ele existir e, caso contrário, retorna um erro 404:

from fastapi import APIRouter, HTTPException, status
from fastapi.responses import FileResponse
from loguru import logger
from .. import config
router = APIRouter()
@router.get('/{token}/{filename}', status_code=status.HTTP_200_OK)
async def download_file(token: str, filename: str) -> FileResponse:
    if not file_exists(token, filename):
        raise HTTPException(status.HTTP_404_NOT_FOUND)
    path = config.UPLOAD_DIR / token / filename
    logger.info(f'Downloading file {token}/{filename}')
    return FileResponse(path)

Para retornar o arquivo, usamos a classe FileResponse do FastAPI (linhas 11 e 16).

DELETE /{token}/{filename}

O próximo endpoint é o de remoção de arquivo. Se o arquivo não existir, a resposta é um código 404. Caso exista, o arquivo é removido, o agendamento de remoção é cancelado e a resposta é um código 204:

from .. import config
from ..file_utils import file_exists, remove_file
from ..resources import scheduler
from fastapi import APIRouter, HTTPException, status
from loguru import logger
router = APIRouter()  # o mesmo router do endpoint anterior
@router.delete('/{token}/{filename}', status_code=status.HTTP_204_NO_CONTENT)
async def delete_file(token: str, filename: str) -> None:
    if not file_exists(token, filename):
        raise HTTPException(status.HTTP_404_NOT_FOUND)
    logger.info(f'Deleting file {token}/{filename}')
    remove_file(token, filename)
    scheduler.remove_job(f'{token}/{filename}')

A remoção do agendamento é baseada no id, que será definida como {token}/{filename} quando o arquivo for salvo e o agendamento for criado.

POST /

O último endpoint é o de upload. No caso em que tudo acontece como o esperado:

  1. O arquivo é salvo
  2. Um agendamento de remoção é criado
  3. O código de retorno é 201
  4. A URL de download do arquivo é retornada como cabeçalho Location e também como resposta da requisição

Se o tamanho do arquivo exceder o limite, o arquivo não é salvo e o código de retorno é 413.

from datetime import datetime, timedelta
from pathlib import Path
from typing import Annotated
from urllib.parse import urljoin
from fastapi import APIRouter, Depends, Header, HTTPException, Request, Response, UploadFile, status
from fastapi.responses import PlainTextResponse
from loguru import logger
from .. import config
from ..file_utils import save_file
from ..resources import scheduler
ContentLength = Annotated[int | None, Header(lte=config.FILE_SIZE_LIMIT)]
router = APIRouter()
@router.post('/', response_class=PlainTextResponse, status_code=status.HTTP_201_CREATED)
async def upload_file(
    file: UploadFile,
    request: Request,
    response: Response,
    content_length: ContentLength = None,  # noqa: ARG001
) -> str:
    """
    Upload a file
    1. In order to prevent name collision, the file is saved in a directory with a random name.
    2. The path to the file is returned as plain text but also set as the Location header.
    """
    if not file.filename:
        raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, detail='Missing filename')
    # save the file
    logger.info(f'Uploading file {file.filename}')
    try:
        token, filename = await save_file(file)
    except OSError:
        raise HTTPException(status.HTTP_413_REQUEST_ENTITY_TOO_LARGE) from None
    # schedule file removal
    scheduler.add_job(
        remove_file,
        'date',
        run_date=datetime.utcnow() + timedelta(seconds=config.TIMEOUT_INTERVAL),
        args=[token, filename],
        id=f'{token}/{filename}',
    )
    # return the URL location of the file
    if request.headers.get('X-Forwarded-Proto'):  # served by a reverse proxy
        host = request.headers.get('X-Forwarded-Host') or request.headers.get('Host')
        location = urljoin(
            f'{request.headers["X-Forwarded-Proto"]}://{host}',
            f'{request.url.path}{router.prefix}{token}/{filename}',
        )
    else:  # served locally (development or testing)
        location = urljoin(request.url._url, f'{router.prefix}/{token}/{filename}')
    response.headers['Location'] = location
    return location

A declaração content_length (linha 23) serve como uma verificação inicial do tamanho do arquivo, baseado no cabeçalho Content-Length. A outra verificação é feita durante o salvamento do arquivo (linha 38).

A URL de download do arquivo (linhas 52 a 59) depende de onde a aplicação está rodando. Se for em produção, haverá um proxy reverso que fornecerá um cabeçalho X-Forwarded-Proto provavelmente contendo https. O domínio é obtido a partir do cabeçalho X-Forwarded-Host, que determina o domínio originalmente usado, ou Host se o primeiro não estiver disponível. Combinando essas informações com o caminho do endpoint, o token e nome do arquivo, temos a URL de download completa.

No segundo caso, a aplicação está rodando localmente (desenvolvimento ou teste) e a URL de download é montada a partir da URL da requisição. A resposta será algo como http://localhost:5000/{token}/{filename}.

Conforme recomendado na documentação do MDN sobre resposta com código 201, vamos retornar a URL do recurso criado tanto no corpo da mensagem de resposta quanto no cabeçalho Location. Para retornar um texto simples, usamos PlainTextResponse (linha 18). Caso contrário, o FastAPI retornaria um JSON com a URL.

Os endpoints estão todos disponíveis no módulo routers/file.py.

Testes dos Endpoints

Agora que os endpoints estão prontos, vamos criar testes para eles no arquivo test_file.py.

Teste do Ciclo de Vida de Arquivos

O primeiro teste cobre todo o ciclo de vida de um arquivo. Vamos criar um arquivo temporário, fazer o upload desse mesmo arquivo duas vezes, baixar o arquivo, remover o arquivo e tentar baixar o arquivo novamente:

from pathlib import Path
from typing import Final
from fastapi import status
from httpx import AsyncClient
from transfer.file_utils import file_exists
from transfer.resources import scheduler
async def test_file_lifecycle(tmp_path: Path, client: AsyncClient) -> None:
    # create a file
    hello_world = tmp_path / 'hello_world.txt'
    text = 'hellow world'
    hello_world.write_text('hello world')
    # upload the file
    with hello_world.open('rb') as f:
        resp = await client.post('/', files={'file': f})
    assert resp.status_code == status.HTTP_201_CREATED
    location = resp.headers['Location']
    token, filename = location.split('/')[-2:]
    assert location == resp.text == f'http://testserver/{token}/{filename}'
    assert file_exists(token, filename)
    job_id = f'{token}/{filename}'
    assert scheduler.get_job(job_id) is not None
    # upload the same file but faking a reverse proxy
    with hello_world.open('rb') as f:
        resp = await client.post(
            '/',
            files={'file': f},
            headers={'X-Forwarded-Proto': 'https', 'X-Forwarded-Host': 'transfer.pronus.xyz'},
        )
    assert resp.status_code == status.HTTP_201_CREATED
    token2, filename2 = resp.text.split('/')[-2:]
    assert token != token2
    assert filename == filename2
    assert (
        resp.headers['Location'] == resp.text == f'https://transfer.pronus.xyz/{token2}/{filename2}'
    )
    assert scheduler.get_job(f'{token2}/{filename2}') is not None
    # get existing file
    resp = await client.get(location)
    assert resp.status_code == status.HTTP_200_OK
    assert resp.text == 'hello world'
    assert scheduler.get_job(job_id) is not None
    # delete file
    resp = await client.delete(f'/{token}/{filename}')
    assert resp.status_code == status.HTTP_204_NO_CONTENT
    assert not file_exists(token, filename)
    assert scheduler.get_job(job_id) is None
    # get non-existing file
    resp = await client.get(location)
    assert resp.status_code == status.HTTP_404_NOT_FOUND
    # delete non-existing file
    resp = await client.delete(f'/{token}/{filename}')
    assert resp.status_code == status.HTTP_404_NOT_FOUND

O teste de upload de um arquivo deve verificar o resultado da requisição, o cabeçalho Location e o corpo da resposta (linhas 20 a 23), se o arquivo foi realmente salvo e se um agendamento foi criado (linhas 24 a 26).

O segundo teste de upload usa o mesmo arquivo, mas simula um proxy reverso (linha 33). A resposta deve conter um token diferente do primeiro upload (linha 36 e 37) e o mesmo nome de arquivo (linha 38). As demais verificações para o upload são as mesmas.

O teste de download verifica se a resposta é bem sucedida, se o conteúdo retornado está correto e se o agendamento ainda existe (linhas 44 a 48).

O teste de remoção verifica se o código de retorno é 204, se o arquivo foi realmente removido e se o agendamento foi cancelado (linhas 50 a 54).

Ao tentar obter ou remover o arquivo novamente, o código de retorno deve ser 404 (linha 56 e 62).

Teste de Limite de Tamanho de Arquivo

Um outro teste diferente será criado para cobrir o caso em que o tamanho do arquivo excede o limite:

from pathlib import Path
from typing import Final
from unittest.mock import patch
from fastapi import status
from httpx import AsyncClient
from transfer import config
from transfer.file_utils import file_exists
from transfer.resources import scheduler
SIZE_LIMIT: Final[int] = 10
@patch('transfer.config.FILE_SIZE_LIMIT', SIZE_LIMIT)
async def test_upload_file_over_size_limit(tmp_path: Path, client: AsyncClient) -> None:
    hello_world = tmp_path / 'over_sized.txt'
    hello_world.write_text('x' * (SIZE_LIMIT + 1))
    with hello_world.open('rb') as f:
        resp = await client.post('/', files={'file': f})
    assert resp.status_code == status.HTTP_413_REQUEST_ENTITY_TOO_LARGE
    # try to cheat by forging the Content-Length header
    with hello_world.open('rb') as f:
        resp = await client.post(
            '/', files={'file': f}, headers={'Content-Length': str(SIZE_LIMIT // 2)}
        )
    assert resp.status_code == status.HTTP_413_REQUEST_ENTITY_TOO_LARGE

O teste começa alterando o tamanho máximo permitido para 10 bytes durante o teste (linha 15). Em seguida, cria um arquivo de 11 bytes (linha 18) e tenta fazer o upload (linha 21). O código de retorno deve ser 413 (linha 22).

Na segunda tentativa, o tamanho do arquivo é forjado para metade do tamanho máximo (linha 27). O resultado deve ser o mesmo (linha 29).

Execução dos Testes

O resultado dos testes é:

(transfer) $ make test
pytest -x --cov-report term-missing --cov-report html --cov-branch \
    --cov transfer/
========================== test session starts =========================
platform linux -- Python 3.11.4, pytest-7.4.3, pluggy-1.3.0
rootdir: /home/andre/projetos/tutoriais/transfer
configfile: pyproject.toml
plugins: alt-pytest-asyncio-0.7.2, anyio-4.0.0, cov-4.1.0
collected 9 items

tests/test_file_utils.py ...                                      [ 33%]
tests/test_logging.py ....                                        [ 77%]
tests/routers/test_file.py ..                                     [100%]

---------- coverage: platform linux, python 3.11.4-final-0 -----------
Name                         Stmts   Miss Branch BrPart  Cover   Missing
------------------------------------------------------------------------
transfer/__init__.py             0      0      0      0   100%
transfer/config.py              20      1      2      1    91%   11
transfer/exception_handlers.py  11      0      0      0   100%
transfer/file_utils.py          37      1     12      0    98%   39
transfer/logging.py             25      0      8      2    94%   15->19, 30->32
transfer/main.py                14      0      2      0   100%
transfer/middleware.py          33      0      4      0   100%
transfer/resources.py           27      0      4      0   100%
transfer/routers/__init__.py     0      0      0      0   100%
transfer/routers/file.py        42      1     16      1    97%   33
------------------------------------------------------------------------
TOTAL                          209      3     48      4    97%
Coverage HTML written to dir htmlcov


=========================== 9 passed in 0.49s ==========================

Teste Manual

Para testar o projeto manualmente, primeiro é necessário por o projeto para rodar:

(transfer)  $ make run
ENV=development docker compose -f docker-compose.yml -f docker-compose.dev.yml up --build
[+] Building 8.8s (16/16) FINISHED
=> [app internal] load .dockerignore                                0.0s
=> => transferring context: 113B                                    0.0s
=> => naming to docker.io/library/transfer                          0.0s
...
[+] Running 2/2
✔ Container transfer  Recreated                                     0.1s
✔ Container caddy     Recreated                                     0.0s
Attaching to caddy, transfer
caddy     | {"level":"info","ts":1698771838.952902,"msg":"using provided configuration","config_file":"/etc/caddy/Caddyfile","config_adapter":"caddyfile"}
...
caddy     | {"level":"info","ts":1698771839.0067704,"msg":"serving initial configuration"}
transfer  | {
transfer  |   "timestamp": "2023-10-31T17:04:01.684941+00:00",
transfer  |   "level": "DEBUG",
transfer  |   "message": "config vars",
transfer  |   "source": "resources.py:show_config:39",
transfer  |   "BUFFER_SIZE": 1048576,
transfer  |   "DEBUG": true,
transfer  |   "ENV": "development",
transfer  |   "FILE_SIZE_LIMIT": 5242880,
transfer  |   "PYGMENTS_STYLE": "github-dark",
transfer  |   "REQUEST_ID_LENGTH": 8,
transfer  |   "TESTING": false,
transfer  |   "TIMEOUT_INTERVAL": 3600,
transfer  |   "TOKEN_LENGTH": 8,
transfer  |   "UPLOAD_DIR": "/tmp/transfer_files"
transfer  | }
transfer  |
transfer  |
transfer  | {
transfer  |   "timestamp": "2023-10-31T17:04:01.687691+00:00",
transfer  |   "level": "INFO",
transfer  |   "message": "started...",
transfer  |   "source": "resources.py:startup:29"
transfer  | }
transfer  |
transfer  |
transfer  | [2023-10-31 17:04:01 +0000] [9] [INFO] Running on http://0.0.0.0:5000 (CTRL + C to quit)

Há duas opções de teste. Uma opção é interagindo com a interface do Swagger, acessível em https://localhost/:

A segunda forma é através da linha de comando, usando curl ou httpie, conforme apresentado anteriormente na seção do detalhamento técnico.

Considerações Finais da Parte 1

Este artigo apresentou um tutorial sobre o desenvolvimento de uma aplicação web para compartilhamento de arquivos. Partimos de uma especificação do projeto e criamos cada parte necessária até chegar a uma aplicação funcional, incluindo os testes automatizados correspondentes. Ainda há alguns detalhes que serão tratados em um próximo artigo, como a criação de uma página estática com instruções de operação e a configuração do proxy reverso para lidar com o limite de uploads por IP.

Um dos objetivos alcançados foi mostrar como criar uma aplicação web com o FastAPI de forma simples e rápida usando o gabarito do projeto FastAPI mínimo como base. Muito tempo e esforço foi economizado porque a estrutura básica da aplicação já veio pronta do gabarito, sendo necessário apenas incluir as novas funcionalidades do projeto.

Artigo anterior: Como Criar Logs Estruturados e Rastreáveis em Aplicações FastAPI

Comentários

Comments powered by Disqus