= Напредно развивање на апликацијата = Прототипот од фазата P4 е надграден со базен на конекции и со експлицитни трансакции. Изворниот код е во git репозиториумот на проектот, во папката cityfix. == Трансакции == Апликацијата користи конекции во режим autocommit, па секоја поединечна команда е посебна трансакција. Таму каде што една операција на корисникот бара повеќе команди кои мора да успеат заедно, командите се извршуваат во експлицитна трансакција со with conn.transaction(): ако која било команда фрли грешка (вклучувајќи ги и грешките од тригерите од фазата P7), целата трансакција автоматски се поништува, а корисникот добива порака. Кај операциите каде двајца корисници можат истовремено да ја менуваат истата пријава, пријавата се заклучува со SELECT ... FOR UPDATE. Втората трансакција чека додека првата не заврши, а потоа ја гледа новата состојба и безбедно се откажува. === Трансакција 1 - Поднесување на пријава (cityfix/citizen.py, UC0003) === Пријавата, првиот запис во историјата на статуси и сите фотографии мора да се зачуваат заедно. Ако внесот на која било фотографија не успее, не смее да остане пријава без фотографиите, ниту пријава без запис во историјата (тоа го бара и одложеното ограничување reports_status_consistency од P7). {{{#!python try: with pool.connection() as conn: with conn.transaction(): report = conn.execute( """ INSERT INTO reports (description, location_text, latitude, longitude, category_id, citizen_id) VALUES (%s, %s, %s, %s, %s, %s) RETURNING report_id, created_at, priority """, (description, location_text, latitude, longitude, category["category_id"], citizen["citizen_id"]), ).fetchone() conn.execute( """ INSERT INTO status_logs (status, changed_at, report_id, worker_id) VALUES ('submitted', %s, %s, NULL) """, (report["created_at"], report["report_id"]), ) with conn.cursor() as cur: cur.executemany( "INSERT INTO photos (image_url, report_id) VALUES (%s, %s)", [(f"/uploads/reports/{report['report_id']}/{p}", report["report_id"]) for p in photos], ) except psycopg.Error as exc: print(f"\n Пријавата не е зачувана: {db.error_message(exc)}") }}} Приоритетот се чита од RETURNING, бидејќи тригерот reports_before_insert (P7) може автоматски да го покачи ако истиот проблем е веќе пријавен во близина. === Трансакција 2 - Конзистентен преглед на пријава (cityfix/citizen.py, UC0004) === Деталите за пријавата се читаат со четири прашалници (пријава, историја, коментари, фотографии). Ако работник го смени статусот помеѓу нив, граѓанинот би можел да види статус кој не се совпаѓа со историјата. Затоа сите четири прашалници се извршуваат во една трансакција со ниво на изолација REPEATABLE READ, само за читање, па гледаат иста слика на базата. {{{#!python with pool.connection() as conn: with conn.transaction(): conn.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY") report = conn.execute("SELECT r.description, r.status, ... WHERE r.report_id = %s", (report_id,)).fetchone() history = conn.execute("SELECT l.status, l.changed_at, ... FROM status_logs l ...", (report_id,)).fetchall() comments = conn.execute("SELECT cm.content, ... FROM comments cm ...", (report_id,)).fetchall() photos = conn.execute("SELECT photo_id, image_url FROM photos WHERE report_id = %s", (report_id,)).fetchall() }}} === Трансакција 3 - Промена на статусот со коментар (cityfix/worker.py, UC0005) === При промена на статусот работникот може да остави и коментар за граѓанинот. Коментарот мора да се внесе пред промената, бидејќи тригерот comments_before_insert (P7) не дозволува коментари на затворена пријава. Ако промената на статусот не успее, и коментарот мора да се поништи. На една пријава може да бидат доделени повеќе работници, па пријавата прво се заклучува и се проверува дека статусот не е променет во меѓувреме. {{{#!python def change_status_tx(pool, worker_id, report_id, expected_status, new_status, note=None, comment=None, hold_seconds=0): with pool.connection() as conn: with conn.transaction(): current = conn.execute( "SELECT status FROM reports WHERE report_id = %s FOR UPDATE", (report_id,), ).fetchone() if hold_seconds: conn.execute("SELECT pg_sleep(%s)", (hold_seconds,)) if current["status"] != expected_status: raise StatusConflict( f"Пријавата во меѓувреме е променета во „{ui.STATUS[current['status']]}“. " "Обиди се повторно.") if comment: conn.execute( "INSERT INTO comments (content, report_id, worker_id) VALUES (%s, %s, %s)", (comment, report_id, worker_id), ) conn.execute("CALL change_report_status(%s, %s, %s, %s)", (report_id, worker_id, new_status, note)) }}} Процедурата change_report_status и тригерите од P7 ги проверуваат дозволените премини и доделеноста на работникот и го усогласуваат статусот во reports. Параметарот hold_seconds се користи само во демонстрацијата подолу, за двете трансакции сигурно да се преклопат. === Трансакција 4 - Доделување на пријава со промена на приоритет (cityfix/admin.py, UC0007) === При доделување, администраторот може да го смени и приоритетот. Двете промени мора да се зачуваат заедно. Двајца администратори можат истовремено да ја гледаат листата на недоделени пријави, па пријавата се заклучува и се проверува дека е сè уште отворена и недоделена. {{{#!python def assign_report(pool, admin_id, report_id, worker_id, priority, note, hold_seconds=0): with pool.connection() as conn: with conn.transaction(): current = conn.execute( "SELECT status, priority FROM reports WHERE report_id = %s FOR UPDATE", (report_id,), ).fetchone() if hold_seconds: conn.execute("SELECT pg_sleep(%s)", (hold_seconds,)) if current["status"] in ("resolved", "rejected"): raise AssignmentConflict("пријавата во меѓувреме е затворена") taken = conn.execute( """ SELECT w.full_name FROM assignments a JOIN workers w ON w.worker_id = a.worker_id WHERE a.report_id = %s """, (report_id,), ).fetchone() if taken: raise AssignmentConflict( f"пријавата во меѓувреме е доделена на {taken['full_name']}") if priority != current["priority"]: conn.execute("UPDATE reports SET priority = %s WHERE report_id = %s", (priority, report_id)) return conn.execute( """ INSERT INTO assignments (note, admin_id, report_id, worker_id) VALUES (%s, %s, %s, %s) RETURNING assignment_id, assigned_at """, (note, admin_id, report_id, worker_id), ).fetchone() }}} == Базен на конекции == === Поставување === Се користи библиотеката psycopg_pool (инсталирана преку requirements.txt со psycopg[binary,pool]). Базенот се креира еднаш при стартување на апликацијата (cityfix/db.py) и се затвора при излез (cityfix/__main__.py). Параметрите се читаат од секцијата [pool] во config.ini: {{{ [pool] min_size = 2 max_size = 10 timeout = 10 }}} {{{#!python def create_pool(): cfg = _config() db = cfg["database"] pool_cfg = cfg["pool"] if cfg.has_section("pool") else {} conninfo = (f"host={db.get('host', 'localhost')} port={db.get('port', '5432')} " f"dbname={db['dbname']} user={db['user']} password={db['password']} " "options='-c search_path=project'") pool = ConnectionPool( conninfo, min_size=int(pool_cfg.get("min_size", 2)), max_size=int(pool_cfg.get("max_size", 10)), timeout=float(pool_cfg.get("timeout", 10)), max_idle=300, kwargs={"row_factory": dict_row, "autocommit": True}, check=ConnectionPool.check_connection, name="cityfix", open=False, ) pool.open(wait=True) return pool }}} * min_size - број на конекции кои се отвораат при стартување и секогаш се чуваат отворени; * max_size - најголем број истовремено отворени конекции; при поголем број барања, базенот отвора дополнителни конекции до оваа граница; * timeout - најдолго време (во секунди) кое барањето чека на слободна конекција, пред да се пријави грешка; * max_idle - конекциите над min_size кои не се користат 5 минути се затвораат; * check - пред секое издавање, конекцијата се проверува, па прекинатите конекции (на пример по рестарт на серверот) автоматски се заменуваат; * kwargs - секоја конекција ги враќа редовите како речници, работи во autocommit режим, а search_path е поставен на шемата project. In main(): {{{#!python pool = db.create_pool() try: ... role[1](pool) finally: pool.close() }}} === Користење на конекција од базенот === Секоја операција зема конекција од базенот со with pool.connection() as conn: и ја враќа веднаш по завршувањето, дури и ако настане грешка. Со тоа ниту една конекција не е зафатена додека корисникот чита или внесува податоци на екранот. За едноставни прашалници се користат помошни функции во cityfix/db.py: {{{#!python def fetch_all(pool, sql, params=None): with pool.connection() as conn: return conn.execute(sql, params).fetchall() }}} Пример од cityfix/admin.py (листа на работници од погледот v_worker_workload од P7): {{{#!python workers = db.fetch_all(pool, """ SELECT worker_id, full_name, active_reports, active_high_priority FROM v_worker_workload ORDER BY active_reports, full_name """) }}} Пример од cityfix/citizen.py (проверка на слични пријави со функцијата од P7): {{{#!python similar = db.fetch_all( pool, "SELECT * FROM find_similar_reports(%s::integer, %s::numeric, %s::numeric, 150)", (category["category_id"], latitude, longitude), ) }}} Трансакциите 1-4 погоре ја користат истата конекција од базенот за сите команди во трансакцијата. == Демонстрација на паралелна работа == Модулот cityfix/demo_concurrency.py ги стартува истовремено повеќе нишки кои го користат истиот базен и истите функции како апликацијата. Се стартува со python -m cityfix.demo_concurrency (демонстрацијата ги менува податоците, па по неа треба повторно да се изврши data_load.sql). Резултат од извршувањето со max_size = 5: {{{ 1) Базен на конекции: 20 истовремени барања [на почеток] отворени конекции: 2, слободни: 2, вкупно барања: 0, чекале на слободна конекција: 0 20 барања завршија за 0.84 s [по барањата] отворени конекции: 5, слободни: 5, вкупно барања: 20, чекале на слободна конекција: 18 2) Двајца администратори истовремено ја доделуваат иста пријава Администратор 1: одбиено - пријавата во меѓувреме е доделена на Горан Трајковски Администратор 2: успешно - внесено доделување бр. 73 3) Двајца доделени работници истовремено го менуваат статусот на иста пријава Работник 1: одбиено - Пријавата во меѓувреме е променета во „Одбиена“. Обиди се повторно. Работник 2: успешно }}} * Во првиот дел, 20 барања од по 0,2 секунди се извршија за 0,84 секунди со само 5 конекции: базенот ги отвори дополнителните конекции до max_size, а останатите барања почекаа слободна конекција наместо да отвораат нови. * Во вториот и третиот дел, двете трансакции се обидуваат истовремено да ја сменат истата пријава. Втората трансакција чека на заклучувањето FOR UPDATE, потоа ја гледа променетата состојба и се откажува, без да внесе ништо. Која од двете трансакции ќе успее зависи од распоредот на нишките, но секогаш успева точно една. == Користење на вештачка интелигенција == [wiki:AdvancedApplicationDevelopmentAIUsage Користење на вештачка интелигенција за напредното развивање на апликацијата]