What I’m trying to do:
I am running a local python script to communicate with my Anvil website using the uplink facility.
I an calling the anvil.server.wait_forever() but want to terminate the server in either of two cases.
case 1. A button on the website to close the anvil.server
case 2. An action caused by the user in the local python code.
What I’ve tried and what’s not working:
For case 1, I have a button which executes: anvil.server.call(‘say_goodbye’, ‘Close local sever’) and in the local script a function @anvil.server.callable
def say_goodbye(message): quit()
but it never quits()
I did try anvil.server.disconnect() - but that created several exceptions.
For case 2, I have a separate thread running in the local thread which is monitoring a local file for a signal to close the script.
However, once anvil.server.wait_forever() is called - this separate thread seems to pause.
I am probably going about this the wrong way. Does anyone know how to fix my method or can suggest a better approach?
Solved.
After inspecting the anvil.server.wait_forever() code, I realised I can create my own version locally which includes a check for “request to close” both locally and from the website button. Also, no need for a separate thread.
Below is a cut-down version of my uplink code. It probably could be done more elegantly, but it works for me.
Another program, with a GUI, will write “CONNECT” as an “action_command” in the interface file, to cause the server to be connected. Additionally, it can write “CLOSE” as an action to close the uplink code.
The anvil website also has a “Close” button, which when clicked calls the “say_goodbye” function which will also close the uplink code.
# anvil_uplink.py
# imports
import time
import os
import json
import anvil.server
# functions
def read_interface_file():
with open(interface_filespec, 'r') as f1:
interface_dict = json.load(f1)
return interface_dict
def check_interface():
global prior_last_modified, interface_dict
last_modified = os.path.getmtime(interface_filespec)
if last_modified != prior_last_modified:
prior_last_modified = last_modified
interface_dict = read_interface_file()
return interface_dict["action_command"]
return ""
def connect_server(anvil_key):
global anvil_connected
print("connecting to server")
anvil.server.connect(anvil_key)
anvil_connected = True
#anvil.server.wait_forever() # use my custom_wait_forever() instead
custom_wait_forever()
def custom_wait_forever():
global keep_anvil_connection
keep_anvil_connection = True
anvil.server._get_connection()
while keep_anvil_connection:
time.sleep(1)
command = check_interface()
if command == "CLOSE":
keep_anvil_connection = False
break
print(f"exiting custom_wait_forever()")
@anvil.server.callable
def say_goodbye(message):
global keep_anvil_connection
print(f"goodbye message from Anvil App: {message}")
keep_anvil_connection = False
if __name__ == "__main__":
print("main running")
anvil_connected = False
keep_anvil_connection = True
prior_last_modified = 0
interface_filespec = r"anvil_uplink_interface.txt"
interface_dict = {
"anvil_key": "SECRET_ANVIL_KEY",
"action_command": ""
}
interface_dict = read_interface_file()
while True:
if not keep_anvil_connection:
print(f"close request received")
break
time.sleep(10)
command = check_interface()
if command == "CLOSE":
keep_anvil_connection = False
print(f"local close request received")
break
elif command == "CONNECT":
if not anvil_connected:
anvil_key = interface_dict["anvil_key"]
connect_server(anvil_key)
print(f"quitting in main")
quit()
rem='''
# example code on anvil website:
def close_button_click(self, **event_args):
try:
anvil.server.call('say_goodbye', 'Close local sever')
self.update_status(Common.OK, f'Closed Local server')
except Exception as err:
self.update_status(Common.ERROR, f'Exception: {err}')
'''
I don’t understand how often you connect here. If you keep connecting, you risk that your uplink is not responsive while connecting. If you only connect once, I don’t understand the logic around the connection.
What do you mean when you say “in the local python code”?
Perhaps you are trying to do two different things with the same script: both listen to the Anvil server with anvil.server.wait_forever() and do something locally.
If that’s what you are trying to do, then I would create two different scripts:
The typical uplink server with anvil.server.wait_forever(), that starts and waits forever.
A second script with your local python code.
Both scripts could kill each other whenever they like using psutil. Here is an untested example from ChatGPT that shows how to kill a Python process. You could put this code in a third module imported by both your scripts:
import psutil
import os
import sys
def kill_process_by_script(script_name):
"""
Finds and kills a running Python process by its script name.
:param script_name: The name of the Python script (e.g., "my_script.py").
"""
current_pid = os.getpid() # Get the PID of the current process
for proc in psutil.process_iter(attrs=['pid', 'name', 'cmdline']):
try:
cmdline = proc.info['cmdline']
if cmdline and script_name in cmdline and proc.info['pid'] != current_pid:
print(f"Killing process {proc.info['pid']} ({cmdline})")
proc.terminate() # Graceful kill
proc.wait(timeout=5) # Wait for termination
return True
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
continue
print(f"No process found running {script_name}")
return False
# Example Usage:
# kill_process_by_script("target_script.py")