@@ -107,7 +107,54 @@ def lint_test(self) -> None:
107
107
self .logger .error ("Resolve errors and exit shell to continue" )
108
108
self .shell ()
109
109
110
+ def find_existing_pr (self ) -> Optional [str ]:
111
+ with contextlib .suppress (
112
+ subprocess .CalledProcessError , json .JSONDecodeError , TypeError
113
+ ):
114
+ for pr in json .loads (
115
+ self .run (
116
+ [
117
+ "gh" ,
118
+ "pr" ,
119
+ "list" ,
120
+ "-H" ,
121
+ self .branch ,
122
+ "-B" ,
123
+ "main" ,
124
+ "--json" ,
125
+ "," .join (("url" , "headRefName" , "baseRefName" )),
126
+ ],
127
+ capture_output = True ,
128
+ check = True ,
129
+ ).stdout .decode ()
130
+ ):
131
+ pr_url = str (pr .pop ("url" ))
132
+ if pr == {"headRefName" : self .branch , "baseRefName" : "main" }:
133
+ return pr_url
134
+ return None
135
+
136
+ def close_existing_pr (self ) -> None :
137
+ # Locate existing PR
138
+ pr_url = self .find_existing_pr ()
139
+ if pr_url :
140
+ if self .dry_run :
141
+ self .logger .info (f"Would close existing PR { pr_url } " )
142
+ else :
143
+ self .run (["gh" , "pr" , "close" , pr_url ])
144
+ self .logger .info (f"Closed existing PR { pr_url } " )
145
+ if self .dry_run :
146
+ return
147
+ # Delete existing branch
148
+ delete_result = self .run (
149
+ ["git" , "push" , "origin" , f":{ self .branch } " ],
150
+ capture_output = True ,
151
+ check = False ,
152
+ )
153
+ if delete_result .returncode == 0 :
154
+ self .logger .info (f"Deleted existing remote branch { self .branch } " )
155
+
110
156
def open_pr (self , message : str ) -> None :
157
+ self .close_existing_pr ()
111
158
if self .dry_run :
112
159
self .logger .success ("Would open PR" )
113
160
return
0 commit comments