return Page(self.backend, res.id, data=res)
def __iter__(self):
- return self._get_pages(
- "SELECT wiki.* FROM wiki_current current \
- LEFT JOIN wiki ON current.id = wiki.id \
- WHERE current.deleted IS FALSE \
- ORDER BY page",
+ return self._get_pages("""
+ SELECT
+ wiki.*
+ FROM
+ wiki_current current
+ LEFT JOIN
+ wiki ON current.id = wiki.id
+ WHERE
+ current.deleted IS FALSE
+ ORDER BY page
+ """,
)
def make_path(self, page, path):
page = Page.sanitise_page_name(page)
# Write page to the database
- page = self._get_page("INSERT INTO wiki(page, author_uid, markdown, changes, address) \
- VALUES(%s, %s, %s, %s, %s) RETURNING *", page, author.uid, content or None, changes, address)
+ page = self._get_page("""
+ INSERT INTO
+ wiki
+ (
+ page,
+ author_uid,
+ markdown,
+ changes,
+ address
+ ) VALUES (
+ %s, %s, %s, %s, %s
+ )
+ RETURNING *
+ """, page, author.uid, content or None, changes, address,
+ )
# Store any linked files
page._store_linked_files()
return ret
def search(self, query, account=None, limit=None):
- res = self._get_pages("SELECT wiki.* FROM wiki_search_index search_index \
- LEFT JOIN wiki ON search_index.wiki_id = wiki.id \
- WHERE search_index.document @@ websearch_to_tsquery('english', %s) \
- ORDER BY ts_rank(search_index.document, websearch_to_tsquery('english', %s)) DESC",
- query, query)
+ res = self._get_pages("""
+ SELECT
+ wiki.*
+ FROM
+ wiki_search_index search_index
+ LEFT JOIN
+ wiki ON search_index.wiki_id = wiki.id
+ WHERE
+ search_index.document @@ websearch_to_tsquery('english', %s)
+ ORDER BY
+ ts_rank(search_index.document, websearch_to_tsquery('english', %s)) DESC
+ """, query, query,
+ )
pages = []
for page in res:
# ACL
def check_acl(self, page, account):
- res = self.db.query("SELECT * FROM wiki_acls \
- WHERE %s ILIKE (path || '%%') ORDER BY LENGTH(path) DESC LIMIT 1", page)
+ res = self.db.query("""
+ SELECT
+ *
+ FROM
+ wiki_acls
+ WHERE
+ %s ILIKE (path || '%%')
+ ORDER BY
+ LENGTH(path) DESC
+ LIMIT 1
+ """, page,
+ )
for row in res:
# Access not permitted when user is not logged in
return File(self.backend, res.id, data=res)
def get_files(self, path):
- files = self._get_files("SELECT * FROM wiki_files \
- WHERE path = %s AND deleted_at IS NULL ORDER BY filename", path)
+ files = self._get_files("""
+ SELECT
+ *
+ FROM
+ wiki_files
+ WHERE
+ path = %s
+ AND
+ deleted_at IS NULL
+ ORDER BY filename
+ """, path,
+ )
return list(files)
if revision:
# Fetch a specific revision
- return self._get_file("SELECT * FROM wiki_files \
- WHERE path = %s AND filename = %s AND created_at <= %s \
- ORDER BY created_at DESC LIMIT 1", path, filename, revision)
+ return self._get_file("""
+ SELECT
+ *
+ FROM
+ wiki_files
+ WHERE
+ path = %s
+ AND
+ filename = %s
+ AND
+ created_at <= %s
+ ORDER BY
+ created_at DESC
+ LIMIT 1
+ """, path, filename, revision,
+ )
# Fetch latest version
- return self._get_file("SELECT * FROM wiki_files \
- WHERE path = %s AND filename = %s AND deleted_at IS NULL",
- path, filename)
+ return self._get_file("""
+ SELECT
+ *
+ FROM
+ wiki_files
+ WHERE
+ path = %s
+ AND
+ filename = %s
+ AND
+ deleted_at IS NULL
+ """, path, filename,
+ )
def get_file_by_path_and_filename(self, path, filename):
- return self._get_file("SELECT * FROM wiki_files \
- WHERE path = %s AND filename = %s AND deleted_at IS NULL",
- path, filename)
+ return self._get_file("""
+ SELECT
+ *
+ FROM
+ wiki_files
+ WHERE
+ path = %s
+ AND
+ filename = %s
+ AND
+ deleted_at IS NULL
+ """, path, filename,
+ )
def upload(self, path, filename, data, mimetype, author, address):
# Replace any existing files
file.delete(author)
# Upload the blob first
- blob = self.db.get("INSERT INTO wiki_blobs(data) VALUES(%s) \
- ON CONFLICT (digest(data, %s)) DO UPDATE SET data = EXCLUDED.data \
- RETURNING id", data, "MD5")
+ blob = self.db.get("""
+ INSERT INTO
+ wiki_blobs(data)
+ VALUES
+ (%s)
+ ON CONFLICT
+ (digest(data, %s))
+ DO UPDATE
+ SET data = EXCLUDED.data
+ RETURNING id
+ """, data, "MD5",
+ )
# Create entry for file
- return self._get_file("INSERT INTO wiki_files(path, filename, author_uid, address, \
- mimetype, blob_id, size) VALUES(%s, %s, %s, %s, %s, %s, %s) RETURNING *", path,
- filename, author.uid, address, mimetype, blob.id, len(data))
+ return self._get_file("""
+ INSERT INTO
+ wiki_files
+ (
+ path,
+ filename,
+ author_uid,
+ address,
+ mimetype,
+ blob_id,
+ size
+ ) VALUES (
+ %s, %s, %s, %s, %s, %s, %s
+ )
+ RETURNING *
+ """, path, filename, author.uid, address, mimetype, blob.id, len(data),
+ )
def render(self, path, text):
return WikiRenderer(self.backend, path, text)