Even if we trust the directory (ie from project default), still have to check the file is in it! TODO make user_id and proejct_string_id optional (ie if directory is supplied?) This needs work but is used in a ton of places so review carefully! Plus want t
(session,
user_id,
project_string_id,
file_id,
directory_id = None,
with_for_update = False,
nowait = False,
skip_locked = False)
| 1036 | |
| 1037 | @staticmethod |
| 1038 | def get_by_id_untrusted(session, |
| 1039 | user_id, |
| 1040 | project_string_id, |
| 1041 | file_id, |
| 1042 | directory_id = None, |
| 1043 | with_for_update = False, |
| 1044 | nowait = False, |
| 1045 | skip_locked = False): |
| 1046 | """ |
| 1047 | |
| 1048 | Even if we trust the directory (ie from project default), |
| 1049 | still have to check the file is in it! |
| 1050 | |
| 1051 | TODO make user_id and proejct_string_id optional (ie if directory is supplied?) |
| 1052 | |
| 1053 | This needs work but is used in a ton of places so review carefully! |
| 1054 | Plus want to keep as "one function" so we don't have a bunch of random checks |
| 1055 | |
| 1056 | """ |
| 1057 | from shared.database.project import Project |
| 1058 | from shared.database.source_control.working_dir import WorkingDirFileLink |
| 1059 | |
| 1060 | if not directory_id: |
| 1061 | project = Project.get(session, project_string_id) |
| 1062 | # start_time = time.time() |
| 1063 | working_dir = project.directory_default |
| 1064 | directory_id = working_dir.id |
| 1065 | |
| 1066 | working_dir_sub_query = session.query(WorkingDirFileLink).filter( |
| 1067 | WorkingDirFileLink.working_dir_id == directory_id).subquery('working_dir_sub_query') |
| 1068 | |
| 1069 | if with_for_update: |
| 1070 | file = session.query(File).with_for_update(nowait = nowait, skip_locked = skip_locked).filter( |
| 1071 | File.id == working_dir_sub_query.c.file_id, |
| 1072 | File.id == file_id).first() |
| 1073 | else: |
| 1074 | file = session.query(File).filter( |
| 1075 | File.id == working_dir_sub_query.c.file_id, |
| 1076 | File.id == file_id).first() |
| 1077 | |
| 1078 | # end_time = time.time() |
| 1079 | # print("File access time", end_time - start_time) |
| 1080 | return file |
| 1081 | |
| 1082 | @staticmethod |
| 1083 | def get_by_id_and_project( |
no test coverage detected