Skip to content

    Chapter 64, Projects

    Intermediate Project: Task CLI

    Here you write the same type of program as the previous project, now as a professional would write: an installable package, with layers, types, tests, and logs. Do this after chapter 41.

    The files for this chapter are in projetos/tarefas/.

    What you will build

    A CLI tarefas with the subcommands adicionar, listar, concluir and remover, installable as a terminal command. The difference from the Basic project is not in what it does, but in how it is organized.

    RequirementConcept that trainsChapter
    Model Tarefadataclass, Self, asdict36, 37
    Removable storage (JSON or memory)Protocol, composition44
    Isolated rules in a serviceLayers, domain exceptions23, 51
    Subcommands with argparseTestable CLI41
    Logs with -vlogging41
    Testing with pytestFixtures, tmp_path, capsys40
    Package with its own commandpyproject.toml, entry points49

    The three-layer architecture

    Quem conhece quem
    cli.py            lê argumentos, imprime, devolve o código de saída
      servico.py      regras: título não vazio, ids sequenciais, tarefa inexistente
        armazenamento.py   JSON em disco, ou memória (para testes)
          modelo.py        a dataclass Tarefa
    

    The service receives the storage in the constructor. Therefore, the service tests use the ArmazenamentoEmMemoria, fast and without touching the disk, and the integration tests use the ArmazenamentoJson with a temporary folder. Switching JSON to SQLite tomorrow requires writing a new class, and nothing else changes.

    The package

    The pyproject.toml declares the command tarefas into [project.scripts]. After installation, tarefas listar works in any folder:

    projetos/tarefas/pyproject.toml
    [build-system]
    requires = ["hatchling"]
    build-backend = "hatchling.build"
    
    [project]
    name = "tarefas"
    version = "0.1.0"
    description = "Gerenciador de tarefas em linha de comando (projeto da parte Intermediário)"
    readme = "README.md"
    requires-python = ">=3.12"
    dependencies = []
    
    [project.scripts]
    tarefas = "tarefas.cli:main"
    
    [dependency-groups]
    dev = ["pytest>=8", "mypy>=1.10", "ruff>=0.6"]
    
    [tool.ruff]
    line-length = 100
    target-version = "py312"
    
    [tool.ruff.lint]
    select = ["E", "F", "I", "B", "UP"]
    
    [tool.mypy]
    python_version = "3.12"
    strict = true
    files = ["src"]
    

    The code

    The model is a dataclass with conversion to and from dictionary:

    projetos/tarefas/src/tarefas/modelo.py
    from dataclasses import asdict, dataclass, field
    from datetime import UTC, datetime
    from typing import Any, Self
    
    
    def agora_iso() -> str:
        return datetime.now(UTC).isoformat(timespec="seconds")
    
    
    @dataclass
    class Tarefa:
        id: int
        titulo: str
        concluida: bool = False
        criada_em: str = field(default_factory=agora_iso)
    
        def para_dict(self) -> dict[str, Any]:
            return asdict(self)
    
        @classmethod
        def de_dict(cls, dados: dict[str, Any]) -> Self:
            return cls(**dados)
    

    The storage is described by a Protocol, with two implementations. Note the salvar of JSON: it writes to a temporary file and then switches, and thus a drop in the middle of the recording never leaves the file corrupted:

    projetos/tarefas/src/tarefas/armazenamento.py
    import json
    from pathlib import Path
    from typing import Protocol
    
    from tarefas.modelo import Tarefa
    
    
    class Armazenamento(Protocol):
        def carregar(self) -> list[Tarefa]: ...
    
        def salvar(self, tarefas: list[Tarefa]) -> None: ...
    
    
    class ArmazenamentoJson:
        def __init__(self, caminho: Path) -> None:
            self._caminho = caminho
    
        def carregar(self) -> list[Tarefa]:
            if not self._caminho.exists():
                return []
            dados = json.loads(self._caminho.read_text(encoding="utf-8"))
            return [Tarefa.de_dict(item) for item in dados]
    
        def salvar(self, tarefas: list[Tarefa]) -> None:
            texto = json.dumps([t.para_dict() for t in tarefas], ensure_ascii=False, indent=2)
            # Grava em um arquivo temporário e troca no final: se o programa cair no meio,
            # o arquivo original continua íntegro.
            temporario = self._caminho.with_suffix(".tmp")
            temporario.write_text(texto, encoding="utf-8")
            temporario.replace(self._caminho)
    
    
    class ArmazenamentoEmMemoria:
        """Usado nos testes: não toca o disco."""
    
        def __init__(self) -> None:
            self._tarefas: list[Tarefa] = []
    
        def carregar(self) -> list[Tarefa]:
            return [Tarefa.de_dict(t.para_dict()) for t in self._tarefas]
    
        def salvar(self, tarefas: list[Tarefa]) -> None:
            self._tarefas = [Tarefa.de_dict(t.para_dict()) for t in tarefas]
    

    The service concentrates the rules and logs what it did:

    projetos/tarefas/src/tarefas/servico.py
    import logging
    
    from tarefas.armazenamento import Armazenamento
    from tarefas.modelo import Tarefa
    
    log = logging.getLogger(__name__)
    
    
    class TarefaNaoEncontrada(Exception):
        def __init__(self, tarefa_id: int) -> None:
            super().__init__(f"tarefa {tarefa_id} não encontrada")
            self.tarefa_id = tarefa_id
    
    
    class ServicoTarefas:
        def __init__(self, armazenamento: Armazenamento) -> None:
            self._armazenamento = armazenamento
    
        def adicionar(self, titulo: str) -> Tarefa:
            titulo = titulo.strip()
            if not titulo:
                raise ValueError("o título não pode ser vazio")
            tarefas = self._armazenamento.carregar()
            proximo_id = max((t.id for t in tarefas), default=0) + 1
            tarefa = Tarefa(id=proximo_id, titulo=titulo)
            self._armazenamento.salvar([*tarefas, tarefa])
            log.info("tarefa %d adicionada", tarefa.id)
            return tarefa
    
        def listar(self, *, somente_pendentes: bool = False) -> list[Tarefa]:
            tarefas = self._armazenamento.carregar()
            return [t for t in tarefas if not t.concluida] if somente_pendentes else tarefas
    
        def concluir(self, tarefa_id: int) -> Tarefa:
            tarefas = self._armazenamento.carregar()
            tarefa = self._buscar(tarefas, tarefa_id)
            tarefa.concluida = True
            self._armazenamento.salvar(tarefas)
            log.info("tarefa %d concluída", tarefa_id)
            return tarefa
    
        def remover(self, tarefa_id: int) -> None:
            tarefas = self._armazenamento.carregar()
            tarefa = self._buscar(tarefas, tarefa_id)
            tarefas.remove(tarefa)
            self._armazenamento.salvar(tarefas)
            log.info("tarefa %d removida", tarefa_id)
    
        @staticmethod
        def _buscar(tarefas: list[Tarefa], tarefa_id: int) -> Tarefa:
            for tarefa in tarefas:
                if tarefa.id == tarefa_id:
                    return tarefa
            raise TarefaNaoEncontrada(tarefa_id)
    

    The CLI only translates between the terminal and the service. The function main receives argv and the storage as parameters (default from chapter 41), and returns a exit code: 0 for success, 1 for user error. The argparse already returns 2 for invalid arguments:

    projetos/tarefas/src/tarefas/cli.py
    import argparse
    import logging
    from pathlib import Path
    
    from tarefas.armazenamento import Armazenamento, ArmazenamentoJson
    from tarefas.servico import ServicoTarefas, TarefaNaoEncontrada
    
    
    def criar_parser() -> argparse.ArgumentParser:
        parser = argparse.ArgumentParser(prog="tarefas", description="Gerenciador de tarefas")
        parser.add_argument("--arquivo", type=Path, default=Path("tarefas.json"))
        parser.add_argument("-v", "--verboso", action="store_true", help="mostra os logs")
        sub = parser.add_subparsers(dest="comando", required=True)
    
        adicionar = sub.add_parser("adicionar", help="cria uma tarefa")
        adicionar.add_argument("titulo")
    
        listar = sub.add_parser("listar", help="mostra as tarefas")
        listar.add_argument("--pendentes", action="store_true")
    
        concluir = sub.add_parser("concluir", help="marca como concluída")
        concluir.add_argument("id", type=int)
    
        remover = sub.add_parser("remover", help="apaga uma tarefa")
        remover.add_argument("id", type=int)
        return parser
    
    
    def main(argv: list[str] | None = None, armazenamento: Armazenamento | None = None) -> int:
        args = criar_parser().parse_args(argv)
        logging.basicConfig(
            level=logging.INFO if args.verboso else logging.WARNING,
            format="%(levelname)s %(name)s: %(message)s",
            force=True,
        )
        servico = ServicoTarefas(armazenamento or ArmazenamentoJson(args.arquivo))
        try:
            if args.comando == "adicionar":
                tarefa = servico.adicionar(args.titulo)
                print(f"Tarefa {tarefa.id} criada: {tarefa.titulo}")
            elif args.comando == "listar":
                tarefas = servico.listar(somente_pendentes=args.pendentes)
                if not tarefas:
                    print("Nenhuma tarefa.")
                for t in tarefas:
                    marca = "x" if t.concluida else " "
                    print(f"[{marca}] {t.id}. {t.titulo}")
            elif args.comando == "concluir":
                print(f"Tarefa {servico.concluir(args.id).id} concluída.")
            elif args.comando == "remover":
                servico.remover(args.id)
                print(f"Tarefa {args.id} removida.")
        except (TarefaNaoEncontrada, ValueError) as erro:
            print(f"Erro: {erro}")
            return 1
        return 0
    
    projetos/tarefas/src/tarefas/__main__.py
    from tarefas.cli import main
    
    raise SystemExit(main())
    

    The tests

    The service tests use in-memory storage and the fixture servico. The JSON storage test confirms the round trip through the disk and that the temporary file does not remain:

    projetos/tarefas/tests/test_servico.py
    import pytest
    
    from tarefas.armazenamento import ArmazenamentoEmMemoria, ArmazenamentoJson
    from tarefas.servico import ServicoTarefas, TarefaNaoEncontrada
    
    
    @pytest.fixture
    def servico() -> ServicoTarefas:
        return ServicoTarefas(ArmazenamentoEmMemoria())
    
    
    def test_adicionar_gera_ids_sequenciais(servico: ServicoTarefas) -> None:
        assert servico.adicionar("a").id == 1
        assert servico.adicionar("b").id == 2
    
    
    def test_titulo_vazio_e_recusado(servico: ServicoTarefas) -> None:
        with pytest.raises(ValueError, match="vazio"):
            servico.adicionar("   ")
    
    
    def test_concluir_e_filtrar_pendentes(servico: ServicoTarefas) -> None:
        servico.adicionar("a")
        servico.adicionar("b")
        servico.concluir(1)
        assert [t.titulo for t in servico.listar(somente_pendentes=True)] == ["b"]
        assert len(servico.listar()) == 2
    
    
    def test_remover_e_nao_reutilizar_apos_remocao_do_ultimo(servico: ServicoTarefas) -> None:
        servico.adicionar("a")
        servico.remover(1)
        assert servico.listar() == []
        with pytest.raises(TarefaNaoEncontrada):
            servico.remover(1)
    
    
    def test_armazenamento_json_ida_e_volta(tmp_path) -> None:  # type: ignore[no-untyped-def]
        caminho = tmp_path / "t.json"
        ServicoTarefas(ArmazenamentoJson(caminho)).adicionar("persistida")
        outra_instancia = ServicoTarefas(ArmazenamentoJson(caminho))
        assert [t.titulo for t in outra_instancia.listar()] == ["persistida"]
        assert not caminho.with_suffix(".tmp").exists()
    

    The CLI tests call main directly, with capsys to capture the output and tmp_path to isolate the file. One of them confirms the exit code 2 of the argparse:

    projetos/tarefas/tests/test_cli.py
    from pathlib import Path
    
    import pytest
    
    from tarefas.cli import main
    
    
    def rodar(arquivo: Path, *argumentos: str) -> int:
        return main(["--arquivo", str(arquivo), *argumentos])
    
    
    def test_fluxo_completo(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
        arquivo = tmp_path / "t.json"
        assert rodar(arquivo, "adicionar", "Estudar Python") == 0
        assert rodar(arquivo, "adicionar", "Escrever testes") == 0
        assert rodar(arquivo, "concluir", "1") == 0
        capsys.readouterr()
        assert rodar(arquivo, "listar") == 0
        saida = capsys.readouterr().out
        assert "[x] 1. Estudar Python" in saida
        assert "[ ] 2. Escrever testes" in saida
    
    
    def test_tarefa_inexistente_devolve_codigo_1(
        tmp_path: Path, capsys: pytest.CaptureFixture[str]
    ) -> None:
        assert rodar(tmp_path / "t.json", "concluir", "99") == 1
        assert "Erro: tarefa 99 não encontrada" in capsys.readouterr().out
    
    
    def test_argumento_invalido_sai_com_codigo_2(tmp_path: Path) -> None:
        with pytest.raises(SystemExit) as codigo:
            rodar(tmp_path / "t.json", "concluir", "abc")
        assert codigo.value.code == 2
    

    Run

    Terminal
    cd projetos/tarefas
    uv sync
    uv run pytest
    uv run mypy
    uv run ruff check .
    
    Output
    ........                                                                 [100%]
    8 passed
    Success: no issues found in 6 source files
    All checks passed!
    

    Using the CLI (the first argument before the subcommand chooses the file):

    Terminal
    uv run tarefas adicionar "Estudar Python"
    uv run tarefas adicionar "Escrever testes"
    uv run tarefas concluir 1
    uv run tarefas listar
    
    Output
    Tarefa 1 criada: Estudar Python
    Tarefa 2 criada: Escrever testes
    Tarefa 1 concluída.
    [x] 1. Estudar Python
    [ ] 2. Escrever testes
    

    With -v, the service shows what it did, and a user error does not generate a traceback, just a message and the code 1:

    Terminal
    uv run tarefas -v adicionar "com log"
    uv run tarefas concluir 99
    
    Output
    INFO tarefas.servico: tarefa 3 adicionada
    Tarefa 3 criada: com log
    Erro: tarefa 99 não encontrada
    

    Challenges

    1. Priority and deadline. Add prioridade and prazo to Tarefa and sort the listing by them. This changes the model and the already saved files: think about how to read an old JSON without these fields (a default value in de_dict).
    2. SQLite Storage. Write ArmazenamentoSqlite with the same interface and run the same service tests against it. If you need to change the service for this, the abstraction was leaking.
    3. Undo. A command desfazer that reverts the last operation.
    4. Publish. Generate the wheel with uv build and install it in a clean environment, as chapter 49 showed.

    What this project teaches that the previous one did not teach

    In the Basic project, everything was in one file and tested by a homemade function. Here each responsibility has a place, dependencies are injected, the program is testable without a terminal, and the command is installable. This is the difference between a script and a package.