FilesExpand file tree

 master

/

isolation_levels.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

64 lines (51 loc) · 1.72 KB

 master

/

isolation_levels.py

Copy path

Top

File metadata and controls

• Code
• Blame

64 lines (51 loc) · 1.72 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

#!/usr/bin/env python

import logging

import sqlite3

import sys

import threading

import time

logging.basicConfig(level=logging.DEBUG,

format='%(asctime)s (%(threadName)-10s) %(message)s',

)

db_filename = 'todo.db'

isolation_level = sys.argv[1]

def writer():

my_name = threading.currentThread().name

logging.debug('connecting')

with sqlite3.connect(db_filename, isolation_level=isolation_level) as conn:

cursor = conn.cursor()

logging.debug('connected')

cursor.execute('UPDATE task SET priority = priority + 1')

logging.debug('changes made')

logging.debug('waiting to synchronize')

ready.wait() # synchronize

logging.debug('PAUSING')

time.sleep(1)

conn.commit()

logging.debug('CHANGES COMMITTED')

return

def reader():

my_name = threading.currentThread().name

with sqlite3.connect(db_filename, isolation_level=isolation_level) as conn:

cursor = conn.cursor()

logging.debug('waiting to synchronize')

ready.wait() # synchronize

logging.debug('wait over')

cursor.execute('SELECT * FROM task')

logging.debug('SELECT EXECUTED')

results = cursor.fetchall()

logging.debug('results fetched')

return

if __name__ == '__main__':

ready = threading.Event()

threads = [

threading.Thread(name='Reader 1', target=reader),

threading.Thread(name='Reader 2', target=reader),

threading.Thread(name='Writer 1', target=writer),

threading.Thread(name='Writer 2', target=writer),

]

[t.start() for t in threads]

time.sleep(1)

logging.debug('setting ready')

ready.set()

[t.join() for t in threads]