FastAPI ( Python ) CRUD extensions

[juca]

Pessoal preciso de ajuda com um problema, ou tentar entender como eu posso implementar isso no FusionPBX.

Fiz uma API com Python ( FastAPI ) que gerencia o PostgreSQL do FusionPBX, API funciona, eu consigo cirar, editar, buscar e apagar ramais no FusionPBX, mas de alguma forma as funcionaliades parece que não se aplicam a API funciona apenas visualmente falando, por exemplo, se eu criar uma Ramal pela API eu consigo me conectar nesse ramal, mas não consigo fazer ligações, ai se eu acessar o painel do FusionPBX e fizer alguma alteração nesse ramal as funcionalidades funcionam.

Queria ver com vocês se alguém sabe alguma forma de usar uma API ( CRUD ) para gerenciar os ramais do FusionPBX, eu fiz uma que funciona mas parece que falta algo a mais para deixar tudo OK.

~~ en:


Guys, I need help with a problem, or trying to understand how I can implement this in FusionPBX.

I created an API with Python (FastAPI) that manages FusionPBX's PostgreSQL, the API works, I can create, edit, search and delete extensions in FusionPBX, but somehow the features don't seem to apply. The API only works visually speaking, so For example, if I create an Extension via the API I can connect to that extension, but I cannot make calls, then if I access the FusionPBX panel and make any changes to that extension, the features work.

I wanted to see if anyone knows any way to use an API (CRUD) to manage FusionPBX extensions, I made one that works but it seems like there's something else missing to make everything OK.


API: https://github.com/josuejuca/api-pbx

 

[DigitalDaz]

Hi Juca, at a guess, even without taken a look at the code, you need to flush the cache. This is what would be happening if you edit an extension in the gui.

rm /var/cache/fusionpbx/*

 

[pbxgeek]

Flushing cache is the first step but there is a lot more happening behind the scenes when you use GUI to create extensions, voicemails, phone numbers, IVRs and such. Some of these create XML files. The bottom line is that managing database through API is not enough to successfully manage FusionPBX. What are you actually trying to achieve here and why are you not using the GUI?

 

[Adrian Fretwell]

I think for extension creating, you may need a hook to create the voicemail record as well. If you are editing an existing extension then you almost certainly will need to flush the cache as pointed out above.

If you are creating your own API, then you may learn a little by looking at the API available in DjangoPBX. We have organisations here in the UK who interact with DjangoPBX only by using the API. Creating domains, extensions etc.

For you convenience I include here, a python code that creates the extension API endpoint and also a screenshot of using the browsable API, note the "Extra Actions" which allow the API to flush the cache when required and also the "perform_create" function calls for the creation of the voicemail record.

class ExtensionViewSet(viewsets.ModelViewSet):
    """
    API endpoint that allows Extensions to be viewed or edited.
    """
    queryset = Extension.objects.all().order_by('domain_id', 'extension')
    serializer_class = ExtensionSerializer
    filter_backends = [DjangoFilterBackend]
    filterset_fields = ['domain_id', 'toll_allow', 'call_group', 'user_context', 'enabled']
    permission_classes = [
        permissions.IsAuthenticated,
        AdminApiAccessPermission,
    ]

    def perform_update(self, serializer):
        serializer.save(updated_by=self.request.user.username)

    def perform_create(self, serializer):
        obj = serializer.save(updated_by=self.request.user.username)
        VoicemailFunctions().create_vm_record(obj, self.request.user.username)

    @action(detail=False)
    def flush_cache_all_directories(self, request):
        i = 0
        qs = Domain.objects.filter(enabled='true')
        for d in qs:
            ClearCache().directory(d.name)
            i += 1
        return Response({'status': 'directory cache flushed', 'count': i})

    @action(detail=True)
    def flush_cache_directory(self, request, pk=None):
        obj = self.get_object()
        ClearCache().directory(obj.domain_id.name)
        return Response({'status': 'directory cache flushed'})