Source code for archivebox.index.sql

__package__ = 'archivebox.index'

import re

from io import StringIO
from pathlib import Path
from typing import List, Tuple, Iterator
from django.db.models import QuerySet
from django.db import transaction

from .schema import Link
from ..util import enforce_types, parse_date
from ..config import (

### Main Links Index

[docs] @enforce_types def parse_sql_main_index(out_dir: Path=OUTPUT_DIR) -> Iterator[Link]: from core.models import Snapshot return ( Link.from_json(page.as_json(*Snapshot.keys)) for page in Snapshot.objects.all() )
[docs] @enforce_types def remove_from_sql_main_index(snapshots: QuerySet, atomic: bool=False, out_dir: Path=OUTPUT_DIR) -> None: if atomic: with transaction.atomic(): return snapshots.delete() return snapshots.delete()
[docs] @enforce_types def write_sql_main_index(links: List[Link], out_dir: Path=OUTPUT_DIR) -> None: for link in links: # with transaction.atomic(): # write_link_to_sql_index(link) write_link_to_sql_index(link)
[docs] @enforce_types def list_migrations(out_dir: Path=OUTPUT_DIR) -> List[Tuple[bool, str]]: from import call_command out = StringIO() call_command("showmigrations", list=True, stdout=out) migrations = [] for line in out.readlines(): if line.strip() and ']' in line: status_str, name_str = line.strip().split(']', 1) is_applied = 'X' in status_str migration_name = name_str.strip() migrations.append((is_applied, migration_name)) return migrations
[docs] @enforce_types def apply_migrations(out_dir: Path=OUTPUT_DIR) -> List[str]: from import call_command null, out = StringIO(), StringIO() call_command("makemigrations", interactive=False, stdout=null) call_command("migrate", interactive=False, stdout=out) return [line.strip() for line in out.readlines() if line.strip()]
[docs] @enforce_types def get_admins(out_dir: Path=OUTPUT_DIR) -> List[str]: from django.contrib.auth.models import User return User.objects.filter(is_superuser=True)