<jupyter_start><jupyter_text>Service Db> Handles sqlite DB access and initialization.<jupyter_code>#export
import sqlite3
from pathlib import Path
import tempfile,shutil,time<jupyter_output><empty_output><jupyter_text>DB init scriptPut this here to avoid having to locate a "schema.sql" file in different environments.This script will- Initialize the database.- Drop any existing data and create empty tables.<jupyter_code>#export
SCHEMA_SQL="""
DROP TABLE IF EXISTS user;
DROP TABLE IF EXISTS post;

CREATE TABLE user (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  username TEXT UNIQUE NOT NULL,
  password TEXT NOT NULL
);

CREATE TABLE post (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  author_id INTEGER NOT NULL,
  created TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  last_updated TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  title TEXT NOT NULL,
  body TEXT NOT NULL,
  status INTEGER NOT NULL DEFAULT 50,
  FOREIGN KEY (author_id) REFERENCES user (id)
);
"""
#export
class ServiceDb:
    
    def __init__(self,database_file):
        self.db = sqlite3.connect(
            database_file, detect_types=sqlite3.PARSE_DECLTYPES
        )
        self.db.row_factory = sqlite3.Row
    
    def read_user_by_id(self,id): 
        return self.db.execute(
            "SELECT * FROM user WHERE id = ?", (id,)
        ).fetchone()
    
    def read_user_by_username(self,username): 
        return self.db.execute(
            "SELECT * FROM user WHERE username = ?", (username,)
        ).fetchone()
    
    def create_user(self,username,password):
        cursor=self.db.cursor()
        cursor.execute(
            "INSERT INTO user (username, password) VALUES (?, ?)",
            (username, password),
        )
        self.db.commit()
        return cursor.lastrowid
        
    def read_posts_by_author_id(self,author_id): 
        return self.db.execute(
            "SELECT p.id, title, body, strftime('%Y-%m-%d %H:%M:%S',created) created, "
            "       strftime('%Y-%m-%d %H:%M:%S',last_updated) last_updated, author_id, username, status"
            "  FROM post p JOIN user u ON p.author_id = u.id"
            " WHERE u.id = ?"
            "   AND p.status > 30"
            " ORDER BY created DESC",
            (author_id,)
        ).fetchall()
    
    def read_post_by_id(self,author_id,id): 
        return self.db.execute(
            "SELECT p.id, title, body, strftime('%Y-%m-%d %H:%M:%S',created) created, "
            "       strftime('%Y-%m-%d %H:%M:%S',last_updated) last_updated, author_id, username, status"
            "  FROM post p JOIN user u ON p.author_id = u.id"
            " WHERE p.author_id = ?"
            "   AND p.id = ?",
            (author_id,id,),
        ).fetchone()
    
    def create_post(self,author_id,title,body):
        cursor=self.db.cursor()
        cursor.execute(
            "INSERT INTO post (author_id, title, body) VALUES (?, ?, ?)",
            (author_id, title, body),
        )
        self.db.commit()
        return cursor.lastrowid
    
    def update_post_by_id(self,author_id,id,keys,values):
        # TODO: use author_id
        self.db.execute(
            'UPDATE post' +
            '   SET last_updated = CURRENT_TIMESTAMP, ' + ','.join([f'{key}=?' for key in keys]) + 
            ' WHERE id = ?', values + [id]
        )
        self.db.commit()
            
    def prepare_posts_file_by_author_id(self,author_id):
        # TODO: implement for DB service
        return None,None
    
    def upload_posts_from_file(self,author_id,file):
        # TODO: implement for DB service
        pass
def _compare(expected,actual):
    "Check values match for all keys in `expected`, which might not be all keys in `actual`"
    for k in expected.keys(): 
        try:
            if expected[k]!=actual[k]: print(k,expected[k],actual[k])
        except Exception as ex:
            print('failed to check',k,ex)
            print('actual.keys()',actual.keys())
        assert expected[k]==actual[k]

temp_path = tempfile.mkdtemp()
try:
    service=ServiceDb(Path(temp_path)/'web_journal_test.sqlite')
    # init
    service.db.executescript(SCHEMA_SQL)
    # user section
    assert service.read_user_by_id(1234) is None
    assert service.read_user_by_username('test.user') is None
    user_id=service.create_user('test.user','<PASSWORD>')
    expected_user=dict(id=user_id,username='test.user',password='<PASSWORD>')
    actual_user=service.read_user_by_username('test.user')
    _compare(expected_user,actual_user)
    assert actual_user==service.read_user_by_id(user_id)
    _compare(expected_user,service.read_user_by_id(user_id))
    # post section
    assert service.read_posts_by_author_id(123)==[]
    assert service.read_posts_by_author_id(user_id)==[]
    assert service.read_post_by_id(user_id,123) is None
    for i in range(3): service.create_post(user_id,f'title{i}','body')
    post_id=service.create_post(user_id,'title','body')
    for i in range(3): service.create_post(user_id,f'title{i}2','body')
    # don't add created to `expected_post` as we don't know what it's value will be
    expected_post=dict(id=post_id,author_id=user_id,title='title',body='body',username='test.user',status=50)
    posts=service.read_posts_by_author_id(user_id)
    assert len(posts)==7
    _compare(expected_post,posts[3])
    assert isinstance(posts[3]['created'],str)
    post=service.read_post_by_id(user_id,post_id)
    assert post==posts[3]
    assert post['status']==50
    assert post!=service.update_post_by_id(user_id,post_id,['status'],[20])
    expected_post['status']=20
    # deleted posts are readable by ID ...
    _compare(expected_post,service.read_post_by_id(user_id,post_id))
    # but are not returned when reading all posts by author
    assert len(service.read_posts_by_author_id(user_id))==6
    
finally:
    service.db.close()
    shutil.rmtree(temp_path)
# test update
temp_path = tempfile.mkdtemp()
try:
    service=ServiceDb(Path(temp_path)/'web_journal_test.sqlite')
    service.db.executescript(SCHEMA_SQL)
    user_id=service.create_user('test.user','<PASSWORD>')
    for i in range(3): service.create_post(user_id,f'title{i}','body')
    post_id=service.create_post(user_id,'title','body')
    for i in range(3): service.create_post(user_id,f'title{i}2','body')
    post=service.read_post_by_id(user_id,post_id)
    assert post['created']==post['last_updated']
    expected_post=dict(id=post_id,author_id=user_id,title='title',body='body',username='test.user',status=50)
    _compare(expected_post,post)
    time.sleep(1) # wait 1s so that last_updated is different
    updated_post=service.update_post_by_id(user_id,post_id,['title','body'],['new title','new-body'])
    expected_post['title']='new title'
    expected_post['body']='new-body'
    post=service.read_post_by_id(user_id,post_id)
    assert post['created']!=post['last_updated']
    _compare(expected_post,post)    
finally:
    service.db.close()
    shutil.rmtree(temp_path)
#export
def before_request(app):
    return ServiceDb(Path(app.config['DATA_DIR'])/'web_journal.sqlite')
#export
def after_request(app,service):
    service.db.close()
#export
def init_service(app):
    print('service.db.init_service')
    service=ServiceDb(Path(app.config['DATA_DIR'])/'web_journal.sqlite')
    try:
        # This will raise an error if the user table has not been created ...
        service.read_user_by_id(0)
    except:
        # ... so we know we need to create the schema
        service.db.executescript(SCHEMA_SQL)
from nbdev.export import notebook2script
notebook2script()<jupyter_output>Converted 00_core.ipynb.
Converted 40a_service_db.ipynb.
Converted 40b_service_filesystem.ipynb.
Converted 50_web_app.ipynb.
Converted 50b_web_auth.ipynb.
Converted 50c_web_blog.ipynb.
Converted index.ipynb.
{'fraction_non_alphanumeric': 0.12265604931929104, 'fraction_numerical': 0.007192396609298741, 'mean_word_length': 3.437037037037037, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '310936', 'license': 'Apache-2.0', 'n_tokens_mistral': 2652, 'n_tokens_neox': 2578, 'n_words': 521, 'path': '40a_service_db.ipynb', 'repo_name': 'pete88b/web_journal', 'size': '7789'}