Coverage for /builds/BuildGrid/buildgrid/buildgrid/server/cas/storage/index/sql_dialect_delegates/postgresqldelegate.py: 100.00%

17 statements  

« prev     ^ index     » next       coverage.py v6.4.1, created at 2022-06-22 21:04 +0000

1# Copyright (C) 2020 Bloomberg LP 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# <http://www.apache.org/licenses/LICENSE-2.0> 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15""" 

16PostgreSQLDelegate 

17================== 

18 

19Extra functionality for the SQL index when using a PostgreSQL backend. 

20 

21""" 

22from datetime import datetime 

23from typing import List, Optional, Tuple 

24 

25from sqlalchemy.dialects.postgresql import insert 

26from sqlalchemy.orm.session import Session as SessionType 

27from sqlalchemy.sql.functions import coalesce 

28 

29from buildgrid._protos.build.bazel.remote.execution.v2.remote_execution_pb2 import Digest 

30from buildgrid.server.persistence.sql.models import IndexEntry 

31 

32 

33class PostgreSQLDelegate: 

34 

35 @staticmethod 

36 def _save_digests_to_index(digest_blob_pairs: List[Tuple[Digest, Optional[bytes]]], 

37 session: SessionType, max_inline_blob_size: int) -> None: 

38 index_table = IndexEntry.__table__ 

39 update_time = datetime.utcnow() 

40 new_rows = [{ 

41 'digest_hash': digest.hash, 

42 'digest_size_bytes': digest.size_bytes, 

43 'accessed_timestamp': update_time, 

44 'inline_blob': (blob if digest.size_bytes <= max_inline_blob_size else None) 

45 } for (digest, blob) in digest_blob_pairs] 

46 

47 base_insert_stmt = insert(index_table).values(new_rows) 

48 

49 update_stmt = base_insert_stmt.on_conflict_do_update( 

50 index_elements=['digest_hash'], 

51 set_={ 

52 'accessed_timestamp': update_time, 

53 'inline_blob': coalesce(base_insert_stmt.excluded.inline_blob, index_table.c.inline_blob) 

54 } 

55 ) 

56 

57 session.execute(update_stmt)