wrpylib/mwdb.py

   1 """This module contains code to make the access of MediaWiki tables
   2 easy. The module uses sqlalchemy to access the database.
   3 """
   4 from sqlalchemy import Table, Column, types, MetaData
   5
   6
   7 def page_table(metadata: MetaData) -> Table:
   8     """Returns the sqlalchemy Table representing the "page" table in MediaWiki.
   9     :param metadata: metadata = sqlalchemy.MetaData()
  10     """
  11     return Table(
  12         "page", metadata,
  13         Column("page_id", types.Integer, primary_key=True),
  14         Column("page_namespace", types.Integer, nullable=False),
  15         Column("page_title", types.String(255, convert_unicode='force'), nullable=False),
  16         Column("page_restrictions", types.String, nullable=False),  # tinyblob NOT NULL
  17         Column("page_is_redirect", types.Integer, nullable=False),
  18         Column("page_is_new", types.Integer, nullable=False),
  19         Column("page_random", types.Float, nullable=False),
  20         Column("page_touched", types.String(14, convert_unicode='force'), nullable=False),
  21         Column("page_links_updated", types.String(14, convert_unicode='force')),
  22         Column("page_latest", types.Integer, nullable=False),
  23         Column("page_len", types.Integer, nullable=False),
  24         Column("page_content_model", types.String(32, convert_unicode='force')),
  25         Column("page_lang", types.String(35, convert_unicode='force')),
  26     )
  27
  28
  29 def revision_table(metadata: MetaData) -> Table:
  30     """Returns the sqlalchemy Table representing the "revision" table in MediaWiki.
  31     :param metadata: metadata = sqlalchemy.MetaData()
  32     """
  33     return Table(
  34         "revision", metadata,
  35         Column("rev_id", types.Integer, nullable=False, primary_key=True),
  36         Column("rev_page", types.Integer, nullable=False, primary_key=True),
  37         Column("rev_comment_id", types.Integer, nullable=False),
  38         Column("rev_actor", types.Integer, nullable=False),
  39         Column("rev_timestamp", types.String(14, convert_unicode='force'), nullable=False),
  40         Column("rev_minor_edit", types.Integer, nullable=False),
  41         Column("rev_deleted", types.Integer, nullable=False),
  42         Column("rev_len", types.Integer, nullable=True),
  43         Column("rev_parent_id", types.Integer, nullable=True),
  44         Column("rev_sha1", types.String(32), nullable=False),
  45     )
  46
  47
  48 def slots_table(metadata: MetaData) -> Table:
  49     """Returns the sqlalchemy Table representing the "slots" table in MediaWiki.
  50     :param metadata: metadata = sqlalchemy.MetaData()
  51     """
  52     return Table(
  53         "slots", metadata,
  54         Column("slot_revision_id", types.Integer, nullable=False, primary_key=True),
  55         Column("slot_role_id", types.Integer, nullable=False, primary_key=True),
  56         Column("slot_content_id", types.Integer, nullable=False),
  57         Column("slot_origin", types.Integer, nullable=False),
  58     )
  59
  60
  61 def content_table(metadata: MetaData) -> Table:
  62     """Returns the sqlalchemy Table representing the "content" table in MediaWiki.
  63     :param metadata: metadata = sqlalchemy.MetaData()
  64     """
  65     return Table(
  66         "content", metadata,
  67         Column("content_id", types.Integer, nullable=False, primary_key=True),
  68         Column("content_size", types.Integer, nullable=False),
  69         Column("content_sha1", types.String(32, convert_unicode='force'), nullable=False),
  70         Column("content_model", types.Integer, nullable=False),
  71         Column("content_address", types.String(255, convert_unicode='force'), nullable=False),
  72     )
  73
  74
  75 def text_table(metadata: MetaData) -> Table:
  76     """Returns the sqlalchemy Table representing the "text" table in MediaWiki.
  77     :param metadata: metadata = sqlalchemy.MetaData()
  78     """
  79     return Table(
  80         "text", metadata,
  81         Column("old_id", types.Integer, primary_key=True),
  82         Column("old_text", types.String(convert_unicode='force')),
  83         Column("old_flags", types.String(convert_unicode='force')),
  84     )
  85
  86
  87 def user_table(metadata: MetaData) -> Table:
  88     """Returns the sqlalchemy Table representing the "user" table in MediaWiki.
  89     :param metadata: metadata = sqlalchemy.MetaData()
  90     """
  91     return Table(
  92         'user', metadata,
  93         Column("user_id", types.Integer, primary_key=True),
  94         Column("user_name", types.String(255, convert_unicode='force'), nullable=False),
  95         Column("user_real_name", types.String(255, convert_unicode='force'), nullable=False),
  96         Column("user_password", types.UnicodeText, nullable=False),
  97         # "user_newpassword"
  98         # "user_newpass_time"
  99         Column("user_email", types.String(convert_unicode='force'), nullable=False),
 100         # "user_touched"
 101         # "user_token"
 102         # "user_email_authenticated"
 103         # "user_email_token"
 104         # "user_email_token_expires"
 105         # "user_registration"
 106         # "user_editcount"
 107     )
 108
 109
 110 def user_groups_table(metadata: MetaData) -> Table:
 111     """
 112     Returns the sqlalchemy Table representing the "user_groups" table in MediaWiki.
 113
 114     :param metadata: metadata = sqlalchemy.MetaData()
 115     """
 116     return Table(
 117         'user_groups', metadata,
 118         Column('ug_user', types.Integer, nullable=False, primary_key=True),
 119         Column('ug_group', types.String(255, convert_unicode='force'), nullable=False, primary_key=True),
 120         Column("ug_expiry", types.String(14, convert_unicode='force')),
 121     )
 122
 123
 124 def categorylinks_table(metadata: MetaData) -> Table:
 125     """Returns the sqlalchemy Table representing the "categorylinks" table in MediaWiki.
 126     :param metadata: metadata = sqlalchemy.MetaData()
 127     """
 128     return Table(
 129         "categorylinks", metadata,
 130         Column("cl_from", types.Integer, nullable=False, primary_key=True),
 131         Column("cl_to", types.String(255, convert_unicode='force'), nullable=False, primary_key=True),
 132         Column("cl_sortkey", types.String(230, convert_unicode='force'), nullable=False),
 133         Column("cl_timestamp", types.DateTime, nullable=False),
 134     )