Coverage for slidge/db/alembic/versions/211345c68f0b_no_upload_attachments_in_message_subdir.py: 68%

1"""Move files from NO_UPLOAD_PATH to NO_UPLOAD_PATH / messages 

2 

3Revision ID: 211345c68f0b 

4Revises: 39ce1636571a 

5Create Date: 2026-08-22 15:25:18.926618 

6 

7""" 

8 

9import logging 

10import shutil 

11from collections.abc import Sequence 

12from pathlib import Path 

13 

14import sqlalchemy as sa 

15from alembic import op 

16 

17from slidge.core import config 

18 

19# revision identifiers, used by Alembic. 

20revision: str = "211345c68f0b" 

21down_revision: str | None = "39ce1636571a" 

22branch_labels: str | Sequence[str] | None = None 

23depends_on: str | Sequence[str] | None = None 

24 

25 

26def upgrade() -> None: 

27 if config.NO_UPLOAD_PATH is None: 

28 log.info( 

29 "NO_UPLOAD_PATH is unset, no need to update local paths of attachments." 

30 ) 

31 return 

32 bind = op.get_bind() 

33 result = bind.execute(sa.text("SELECT url FROM attachment")) 

34 urls = [row[0] for row in result] 

35 root = Path(config.NO_UPLOAD_PATH) 

36 for url in urls: 

37 suffix = url.removeprefix(config.NO_UPLOAD_URL_PREFIX) 

38 _move(root, suffix) 

39 op.execute( 

40 sa.text(_REPLACE.format(old=config.NO_UPLOAD_URL_PREFIX, new=_get_new())) 

41 ) 

42 

43 

44def downgrade() -> None: 

45 raise RuntimeError("downgrades are not supported") 

46 

47 

48def _get_new() -> str: 

49 assert config.NO_UPLOAD_URL_PREFIX 

50 return config.NO_UPLOAD_URL_PREFIX + "/message" 

51 

52 

53def _move(root: Path, suffix: str) -> None: 

54 src = root / suffix 

55 if not src.exists(): 

56 log.warning("%s does not exist, can't migrate", src) 

57 return 

58 dest = root / "message" / suffix 

59 dest.parent.mkdir(parents=True, exist_ok=True) 

60 shutil.move(src, dest) 

61 

62 

63_REPLACE = "UPDATE attachment SET url = REPLACE(url, '{old}', '{new}')" 

64 

65log = logging.getLogger(__name__)