Forem: Paurakh Sharma Humagain

Using Firebase Cloud Functions to Generate URL Preview

Paurakh Sharma Humagain — Sun, 12 Sep 2021 03:28:44 +0000

Using Firebase Cloud Functions to Generate URL Preview

Hello Flutter Devs,

In this article we are going to learn how to use Firebase cloud functions
to parse the URL og tags and display the URL previews
in the Flutter app.

How this works?

User adds a url in the Firestore collection urls.
Cloud functions then triggers onCreate() which:

i) Makes the request to the url

ii) Parses the og tags from the response

iii) Saves the og details in the Firestore collection previews

Let's get started

We are going to use FloatingActionButton to open the dialog where user can add new url.

FloatingActionButton(
  onPressed: () async {
    final url = await _showUrlDialog(context);
    if (url == null) return;
    final collection = FirebaseFirestore.instance.collection('urls');
    collection.add({'url': url});

    ScaffoldMessenger.of(context).showSnackBar(
      SnackBar(
        content: Text('URL added successfully!'),
      ),
    );
  },
  child: Icon(Icons.add),
)

Lets implement _shorUrlDialog

Future<String?> _showUrlDialog(BuildContext context) async {
  String url = '';
  final result = await showDialog<String>(
    context: context,
    barrierDismissible: false,
    builder: (BuildContext context) {
      return AlertDialog(
        title: Text('Add URL'),
        content: SizedBox(
          width: double.maxFinite,
          child: TextField(
            onChanged: (value) {
              url = value;
            },
            decoration: InputDecoration(
              hintText: 'https://www.flutter.dev',
              border: OutlineInputBorder(),
            ),
          ),
        ),
        actions: <Widget>[
          TextButton(
            child: Text(
              'Cancel',
            ),
            onPressed: () {
              Navigator.of(context).pop();
            },
          ),
          ElevatedButton(
            child: Text(
              'Add',
            ),
            onPressed: () {
              Navigator.of(context).pop(url);
            },
          ),
        ],
      );
    },
  );

  return result;
}

Now, Initialize a cloud functions for your Firebase project.

Go to the functions directory.

We need to install two packages which makes requesting url and parsing the og tags easier.

npm install got cheerio

Now you are ready to add the cloud functions.

Add this code to your functions/index.js

// packages to make api request and parse response
const got = require('got');
const cheerio = require('cheerio');

const functions = require('firebase-functions');
const admin = require('firebase-admin'); // Cloud Firestore access
const db = admin.initializeApp().firestore();

exports.addUrlPreview = functions.firestore
  .document('urls/{urlId}')
  .onCreate(async (snapshot) => {
    var querySnapshot = await snapshot.ref.get();
    var data = querySnapshot.data();
    var url = data.url;
    if (url.trim() === '') {
      console.log('Empty string... Skipping...');
      return;
    }

    const ogData = await getOGData(url);
    console.log(ogData);

    const publicationRef = db.collection('previews');
    await publicationRef.add(ogData);
  });

/**
 * Get the og details from the URL
 *
 * @param {String} url
 * @return {Promise} of object containing site's title, description, siteName, and image.
 */
async function getOGData(url) {
  if (!url || url.length === 0) return;
  const response = await got(url);

  const $ = cheerio.load(response.body);

  const titleAttr = $('meta[property="og:title"]').attr();
  const descriptionAttr = $('meta[property="og:description"]').attr();
  const siteNameAttr = $('meta[property="og:site_name"]').attr();
  const imageAttr = $('meta[property="og:image"]').attr();

  return {
    title: titleAttr ? titleAttr.content : '',
    description: descriptionAttr ? descriptionAttr.content : '',
    siteName: siteNameAttr ? siteNameAttr.content : '',
    image: imageAttr ? imageAttr.content : '',
    url: url,
  };
}

Now you can display the URL previews in your Flutter app.

I am going to use StreamBuilder for this purpose.

StreamBuilder(
  stream: FirebaseFirestore.instance.collection('urls').snapshots(),
  builder: (context, AsyncSnapshot<QuerySnapshot> snapshot) {
    if (!snapshot.hasData) return Text('Loading..');

    final docs = snapshot.data!.docs;

    if (docs.isEmpty) return Text('Add urls to load their previews');

    final previews =
        docs.map((doc) => doc.data() as Map<String, dynamic>).toList();

    return ListView.separated(
      padding: const EdgeInsets.all(0),
      itemCount: previews.length,
      separatorBuilder: (context, index) => const Divider(),
      itemBuilder: (context, index) {
        final preview = previews[index];
        return ListTile(
          contentPadding: const EdgeInsets.all(0),
          leading: preview['image'].isEmpty
              ? Container(
                  height: 100,
                  width: 100,
                  color: Colors.grey,
                )
              : Image.network(
                  preview['image'],
                  height: 100,
                  width: 100,
                ),
          title: Text(preview['title']),
          subtitle: Column(
            crossAxisAlignment: CrossAxisAlignment.start,
            children: [
              SizedBox(height: 4),
              Text(preview['description']),
              SizedBox(height: 2),
              Text(
                preview['siteName'],
                style: TextStyle(
                  fontWeight: FontWeight.bold,
                ),
              ),
            ],
          ),
        );
      },
    );
  },
)

And you are done.

You can make further improvements to getOGData() method to make it more robust.

You can find the complete source code here: https://github.com/paurakhsharma/cloud_functions_url_preview

Happy Coding!

Floating Audio Player in Flutter

Paurakh Sharma Humagain — Wed, 11 Nov 2020 16:23:33 +0000

Floating Audio Player in Flutter

Today, we are going to learn how to build a floating audio player in Flutter. This method is not limited to the audio player but can use used for the video player as well.

First of all, let's create a new flutter project.

flutter create floating_audio_player

Replace all the codes in main.dart with the following:

import 'package:flutter/material.dart';

void main() {
  runApp(MyApp());
}

class MyApp extends StatelessWidget {
  // This widget is the root of your application.
  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      title: 'Flutter Demo',
      debugShowCheckedModeBanner: false,
      theme: ThemeData(
        primarySwatch: Colors.teal,
        visualDensity: VisualDensity.adaptivePlatformDensity,
      ),
      home: HomePage(),
    );
  }
}

class HomePage extends StatefulWidget {
  @override
  _HomePageState createState() => _HomePageState();
}

class _HomePageState extends State<HomePage> {
  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: Text('Floating media player')),
      body: ListView.builder(itemBuilder: (context, index) {
        return Card(
          child: Container(
            height: 200,
            child: Center(
                child: Text(
              'Item $index',
              style: TextStyle(
                color: index.isEven ? Colors.white : Colors.green,
              ),
            )),
          ),
          color: index.isEven ? Colors.green : Colors.white70,
        );
      }),
    );
  }
}

Here we are rendering an infinite list with cards inside for demonstration purposes only.

Now we are going to create a FAB (Floating Action Button) which can transform into a container. We are going to use AnimationSwitcher for a slick animation.

Update your main.dart and add floatingActionButton

import 'package:flutter/material.dart';

void main() {
  runApp(MyApp());
}

class MyApp extends StatelessWidget {
  // This widget is the root of your application.
  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      title: 'Flutter Demo',
      debugShowCheckedModeBanner: false,
      theme: ThemeData(
        primarySwatch: Colors.teal,
        visualDensity: VisualDensity.adaptivePlatformDensity,
      ),
      home: HomePage(),
    );
  }
}

class HomePage extends StatefulWidget {
  @override
  _HomePageState createState() => _HomePageState();
}

class _HomePageState extends State<HomePage> {
  bool isPlayerOpened = false;
  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: Text('Floating media player')),
      body: ListView.builder(itemBuilder: (context, index) {
        return Card(
          child: Container(
            height: 200,
            child: Center(
                child: Text(
              'Item $index',
              style: TextStyle(
                color: index.isEven ? Colors.white : Colors.green,
              ),
            )),
          ),
          color: index.isEven ? Colors.green : Colors.white70,
        );
      }),
      floatingActionButton: floatingActionItem,
    );
  }

  get floatingActionItem {
    Widget floatingPlayer = GestureDetector(
      child: Row(
        mainAxisAlignment: MainAxisAlignment.center,
        children: [
          SizedBox(width: 35,),
          Container(
            height: 125,
            width: 325,
            color: Colors.teal,
          ),
        ],
      ),
      onTap: () {
        setState(() {
          isPlayerOpened = false;
        });
      },
    );

    Widget floatingActionButton = FloatingActionButton(
      onPressed: () {
        setState(() {
          isPlayerOpened = true;
        });
      },
      child: Icon(Icons.play_arrow_outlined),
    );

    return AnimatedSwitcher(
      reverseDuration: Duration(milliseconds: 0),
      duration: const Duration(milliseconds: 200),
      transitionBuilder: (Widget child, Animation<double> animation) {
        return ScaleTransition(child: child, scale: animation);
      },
      child: isPlayerOpened ? floatingPlayer : floatingActionButton,
    );
  }
}

A bool isPlayerOpened is used to show a play button or a container using an AnimatedSwitcher.

Let's move the floating audio player code into FloatingAudioPlayer.dart, add the following code inside it to show a player UI.

// floating_audio_player.dart

import 'package:flutter/material.dart';

class FloatingAudioPlayer extends StatelessWidget {
  @override
  Widget build(BuildContext context) {
    return Row(
      mainAxisAlignment: MainAxisAlignment.center,
      crossAxisAlignment: CrossAxisAlignment.center,
      children: [
        SizedBox(width: 35),
        Container(
            height: 75,
            width: 325,
            decoration: BoxDecoration(
              color: Colors.teal.withOpacity(0.8),
              borderRadius: BorderRadius.circular(10.0),
            ),
            child: Row(
              children: [
                pauseButton(),
                Expanded(child: progressSlider()),
              ],
            )),
      ],
    );
  }

  Widget pauseButton() {
    return IconButton(
      icon: Icon(Icons.pause_outlined),
      onPressed: pause,
    );
  }

  pause() {}

  Widget progressSlider() {
    const textStyle = TextStyle(color: Colors.white);
    return Column(
      mainAxisSize: MainAxisSize.min,
      children: [
        Padding(
          padding: const EdgeInsets.symmetric(horizontal: 25.0),
          child: Row(
            mainAxisAlignment: MainAxisAlignment.spaceBetween,
            children: [
              Text('00:00', style: textStyle),
              Text('10:10', style: textStyle),
            ],
          ),
        ),
        SizedBox(
          width: 300,
          height: 25,
          child: Slider(
            min: 0,
            max: 280,
            value: 150,
            activeColor: Colors.white,
            inactiveColor: Colors.grey,
            onChanged: (value) {},
          ),
        )
      ],
    );
  }
}

Make sure you replace child inside GestureDetector() in main.dart with newly created FloatingAudioPlayer().

Finally, we are going to implement playing audio. For this I am using just_audio, you can use your favorite audio player.

Add just_audio: ^0.5.6 in your pubspec.yml's dependencies.

Note: Make sure you restart your app after installing dependency which uses platform channels.

For iOS: Make sure you add the following inside your Info.plist file. Even though we do not use the Microphone, iOS requires us to add this.

<key>NSMicrophoneUsageDescription</key>
<string>... explain why you use (or don't use) the microphone ...</string>

Update floating_audio_player.dart and all audio playing functionality.

// floating_audio_player.dart

import 'package:flutter/material.dart';
import 'package:just_audio/just_audio.dart';

class FloatingAudioPlayer extends StatefulWidget {
  final Function onPaused;
  FloatingAudioPlayer({@required this.onPaused});
  @override
  _FloatingAudioPlayerState createState() => _FloatingAudioPlayerState();
}

class _FloatingAudioPlayerState extends State<FloatingAudioPlayer> {
  Duration _duration;
  AudioPlayer _audioPlayer;
  @override
  Widget build(BuildContext context) {
    return Row(
      mainAxisAlignment: MainAxisAlignment.center,
      crossAxisAlignment: CrossAxisAlignment.center,
      children: [
        SizedBox(width: 35),
        Container(
          height: 75,
          width: 325,
          decoration: BoxDecoration(
            color: Colors.teal.withOpacity(0.8),
            borderRadius: BorderRadius.circular(10.0),
          ),
          child: StreamBuilder(
            stream: _audioPlayer.positionStream,
            builder: (context, snapshot) {
              if (snapshot.hasData && _duration != null) {
                if (_audioPlayer.processingState == ProcessingState.completed) {
                  _audioPlayer.stop();
                  Future.delayed(Duration(seconds: 0), widget.onPaused);
                }
                return Row(
                  children: [
                    if (_audioPlayer.playerState.playing)
                      pauseButton()
                    else
                      playButton(),
                    Expanded(child: progressSlider(snapshot.data)),
                  ],
                );
              }

              return Center(child: CircularProgressIndicator());
            },
          ),
        ),
      ],
    );
  }

  Widget playButton() {
    return IconButton(
      icon: Icon(Icons.play_arrow_outlined),
      onPressed: playAudio,
      iconSize: 34,
      color: Colors.white,
    );
  }

  Widget pauseButton() {
    return IconButton(
      icon: Icon(Icons.pause_outlined),
      onPressed: pause,
      iconSize: 34,
      color: Colors.white,
    );
  }

  pause() {
    _audioPlayer.pause();
    Future.delayed(Duration(milliseconds: 500), widget.onPaused);
  }

  Widget progressSlider(Duration position) {
    const textStyle = TextStyle(color: Colors.white);
    return Column(
      mainAxisSize: MainAxisSize.min,
      children: [
        Padding(
          padding: const EdgeInsets.symmetric(horizontal: 25.0),
          child: Row(
            mainAxisAlignment: MainAxisAlignment.spaceBetween,
            children: [
              Text(position.toString().substring(2, 7), style: textStyle),
              Text(_duration.toString().substring(2, 7), style: textStyle),
            ],
          ),
        ),
        SizedBox(
          width: 300,
          height: 25,
          child: Slider(
            min: 0,
            max: _duration.inMilliseconds.toDouble(),
            value: position.inMilliseconds.toDouble(),
            activeColor: Colors.white,
            inactiveColor: Colors.grey,
            onChanged: (value) {
              _audioPlayer.seek(Duration(milliseconds: value.floor()));
            },
          ),
        )
      ],
    );
  }

  @override
  void initState() {
    super.initState();
    initAudio();
  }

  @override
  void dispose() {
    super.dispose();
    _audioPlayer?.dispose();
  }

  initAudio() async {
    _audioPlayer = AudioPlayer();
    final duration = await _audioPlayer.setUrl(
      'https://s3.amazonaws.com/scifri-episodes/scifri20181123-episode.mp3',
    );

    setState(() {
      _duration = duration;
    });

    playAudio();
  }

  playAudio() {
    _audioPlayer.play();
  }
}

Here onPaused is passed inside the FloatingAudioPlayer() widget so that when the user pauses the audio or the audio is played completely, the floating audio player is closed.

Likewise, we need to update our main.dart to implement onPaused

// main.dart

import 'package:flutter/material.dart';

import 'floating_audio_player.dart';

void main() {
  runApp(MyApp());
}

class MyApp extends StatelessWidget {
  // This widget is the root of your application.
  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      title: 'Flutter Demo',
      debugShowCheckedModeBanner: false,
      theme: ThemeData(
        primarySwatch: Colors.teal,
        visualDensity: VisualDensity.adaptivePlatformDensity,
      ),
      home: HomePage(),
    );
  }
}

class HomePage extends StatefulWidget {
  @override
  _HomePageState createState() => _HomePageState();
}

class _HomePageState extends State<HomePage> {
  bool isPlayerOpened = false;
  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: Text('Floating media player')),
      body: ListView.builder(itemBuilder: (context, index) {
        return Card(
          child: Container(
            height: 200,
            child: Center(
                child: Text(
              'Item $index',
              style: TextStyle(
                color: index.isEven ? Colors.white : Colors.green,
              ),
            )),
          ),
          color: index.isEven ? Colors.green : Colors.white70,
        );
      }),
      floatingActionButton: floatingActionItem,
    );
  }

  onPaused() {
    setState(() {
      isPlayerOpened = false;
    });
  }

  get floatingActionItem {
    Widget floatingPlayer = FloatingAudioPlayer(onPaused: onPaused);

    Widget floatingActionButton = FloatingActionButton(
      onPressed: () {
        setState(() {
          isPlayerOpened = true;
        });
      },
      child: Icon(Icons.play_arrow_outlined),
    );

    return AnimatedSwitcher(
      reverseDuration: Duration(milliseconds: 0),
      duration: const Duration(milliseconds: 200),
      transitionBuilder: (Widget child, Animation<double> animation) {
        return ScaleTransition(child: child, scale: animation);
      },
      child: isPlayerOpened ? floatingPlayer : floatingActionButton,
    );
  }
}

That's it 🎉🥳, we have created a floating audio player. As mentioned earlier this is not limited to the audio player but can be used for anything.

Now go and create awesome stuff with it.

You can find the source code here

If you have any suggestions or comments feel free to write down in the comment section, follow me on Twitter @PaurakhSharma for new contents or to suggest one ;)

Happy Coding!

Extend your Flutter code

Paurakh Sharma Humagain — Sun, 06 Sep 2020 12:10:00 +0000

Extend your Flutter code

Are you looking for an alternative to the nested widget structure of Flutter? Extracting the complex widgets into a different widget, can be a good solution. There is another solution, or even both solutions can work together.

For this, we can leverage the power of Dart feature called Extension which was introduced in Dart 2.7.

So, what is `extension` in Dart?

An extension is a way of adding additional functionality to your dart data types.

Let's take the String data type as an example.


String myString = 'Hello world!'

// here toUpperCase() is a method which can
// be applied to a String data type.
print(myString.toUpperCase())

Likewise, we can add our own methods to the data type.

We can create an extension for a datatype using the syntax:

extension StringExtend on String {
    // define methods here
}

And inside this, we can define the methods.


extension StringExtend on String {
  count(String a) {
    int count = 0;
    for (int i = 0; i < this.length; i++) {
      if (this[i] == a) count = count + 1;
    }
    return count;
  }
}

Note: the variable in which the extension method is called can be accessed inside the method using this keyword

Here we have defined a simple method count() which takes a String as a parameter and returns the number occurrence of this string in the called String.

The whole code would look like this:


extension StringExtend on String {
  count(String a) {
    int count = 0;
    for (int i = 0; i < this.length; i++) {
      if (this[i] == a) count = count + 1;
    }
    return count;
  }
}

void main() {
  String myText = 'This is my string';

  // prints 3 in the console
  print(myText.count('i'));
}

Similarly, we can add other methods to the extension.

extension StringExtend on String {
  count(String a) {
    int count = 0;
    for (int i = 0; i < this.length; i++) {
      if (this[i] == a) count = count + 1;
    }
    return count;
  }

  makeFunky() {
    String result = '';
    for (int i = 0; i < this.length; i++) {
      final text = i % 2 == 0 ? this[i].toLowerCase() : this[i].toUpperCase();
      result = result + text;
    }
    return result;
  }
}

void main() {
  String myText = 'This is my string';

  // prints 3 in the console
  print(myText.count('i'));
  // prints iHiS Is mY StTiNg
  print(myText.makeFunky());
}

Now we have a good understanding of the syntax of extension let's see how we can implement this in FLutter.

Since, everything is Widget in flutter. We can add extension to the Widget datatype. Or even an specific Widget such as Text() or Container().

Let's create a flutter extension that changes the font size of a Text() widget.

Without extension the code would look something like this:

build() {
    return Text(
       'Hello world!',
       style: TextStyle(
         fontSize: 24,
        ),
    );
}

With extension

// first define the extension
extension TextModifier on Text {
  Text fontSize(double fontSize) {
    // if the Text() widget doesn't already have an style
    // just apply the fontSize
    if (this.style == null) {
      return Text(
        this.data,
        style: TextStyle(fontSize: fontSize),
      );
    }

    // if there is already an style applied to it
    // extend that style with fontSize
    return Text(
      this.data,
      style: this.style.copyWith(fontSize: fontSize),
    );
  }
}


// actual code implementation

build() {
    return Text('Hello World!').fontSize(24);
}

SEE how readable your code looks like?
Yes there is more boilerplate to define the extension, but that is something you have to do only once and only use it's fruits latter on while using the Text() widget.

Similarly, we can add methods to the extension to chain the methods.

// define extension

extension TextModifier on Text {
  Text color(Color color) {
    if (this.style == null) {
      return Text(
        this.data,
        style: TextStyle(color: color),
      );
    }
    return Text(
      this.data,
      style: this.style.copyWith(color: color),
    );
  }

  Text fontSize(double fontSize) {
    if (this.style == null) {
      return Text(
        this.data,
        style: TextStyle(fontSize: fontSize),
      );
    }

    return Text(
      this.data,
      style: this.style.copyWith(fontSize: fontSize),
    );
  }
}

// use the goodness
build() {
    return Text('Hello World!')
    .fontSize(24)
    .color(Colors.green);
}

SEE how readable the code looks like.

Let's see more examples

Define the extension

extension TextModifier on Text {
  Text color(Color color) {
    if (this.style == null) {
      return Text(
        this.data,
        style: TextStyle(color: color),
      );
    }
    return Text(
      this.data,
      style: this.style.copyWith(color: color),
    );
  }

  Text fontSize(double fontSize) {
    if (this.style == null) {
      return Text(
        this.data,
        style: TextStyle(fontSize: fontSize),
      );
    }

    return Text(
      this.data,
      style: this.style.copyWith(fontSize: fontSize),
    );
  }
}

extension WidgetModifier on Widget {
  Widget padding({horizontal, vertical}) {
    return Padding(
      padding: EdgeInsets.symmetric(
        horizontal: horizontal,
        vertical: vertical,
      ),
      child: this,
    );
  }

  Widget background(Color color) {
    return DecoratedBox(
      decoration: BoxDecoration(
        color: color,
      ),
      child: this,
    );
  }
}

Without extension

build(BuildContext context) {
    return Container(
      color: Colors.purple,
      padding: const EdgeInsets.symmetric(
        horizontal: 10,
        vertical: 20,
      ),
      child: Container(
        color: Colors.lightBlue,
        child: Text(
          'Hello, World!',
          style: TextStyle(
              color: Colors.purple,
              fontSize: 50.0,
            ),
        ),
      ),
    );
  }

With extension

build(BuildContext context) {
    return Text('Hello, World!')
        .color(Colors.purple)
        .fontSize(50.0)
        .background(Colors.lightBlue)
        .padding(horizontal: 10, vertical: 20)
        .background(Colors.purple)
  }

Conclusion

The example app using things taught in this blog can be found at Github

Dart extension can be used in flutter to decrease the boilerplate code for repeating simple widgets. If the widget is more complex it is better to extract it into a separate widget.

I would love to hear your thoughts on using extensions on Flutter code. If you like this you might love an awesome UI framework VelocityX by The Pawan Kumar

Please let me know what you want me to cover next, in the mean time follow me on Twitter

Happy Coding

Beginner Python tips Day - 07 Counter

Paurakh Sharma Humagain — Wed, 10 Jun 2020 14:34:09 +0000


# Python Day -07 - Counter

# useful dict subclass for counting it's elements
# element can be a string, dictionary, or keyword args

from collections import Counter

string_counter = Counter('aasddhhssdhdhdd')

print(string_counter)
# Prints Counter({'d': 6, 'h': 4, 's': 3, 'a': 2})

print(string_counter.most_common(2))
# Prints [('d', 6), ('h', 4)], the most common elements
# with element and element count tuple

print(list(string_counter.elements()))
# Prints ['a', 'a', 's', 's', 's', 'd', 'd', 'd', 'd', 'd', 'd', 'h', 'h', 'h', 'h']
# all the elements

print(string_counter['d'])
# Prints 6, count of element d

print(string_counter['w'])
# Prints 0 as there is no element w

Beginner Python tips Day - 06 try...except...else...finally

Paurakh Sharma Humagain — Tue, 09 Jun 2020 15:32:33 +0000

# Python Day 06 - try...except...else...finally

# try, except(also known as catch) and finally pattern is seen in many programming language
# but on top of that Python comes with else block as well,
# else block runs before finally block
# when there is no exception in the try block

# Example 1

try:
    print(100/0)
    # Doesn't print anything because there is an exception
except ZeroDivisionError:
    print('This does run as there is ZeroDivisionError')
else:
    print("This doesn't run as there is an exception in try block")
finally:
    print('This runs regardless of the exception')

# Example 2

try:
    print(100/5)
    # Prints 20.0
except ZeroDivisionError:
    print("This doesn't run as there is no ZeroDivisionError")
else:
    print("This runs as there is no exception in try block")
finally:
    print('This runs regardless of the exception')

Beginner Python tips Day - 05 while...else, for...else

Paurakh Sharma Humagain — Mon, 08 Jun 2020 14:55:11 +0000

# Python Day 05 - while....else, for...else

# else block in while and for loop is only executed
# if break statement is not executed inside for loop

my_list = ['Python', 'JavaScript', 'Dart', 'Golang', 'Rust']

# while else
index = 0
while index < len(my_list):
    if my_list[index] == 'PHP':
        result = 'Found PHP'
        break
    index += 1
else:
    result = 'PHP not found'

print(result)
# Prints 'PHP not found'
# else block in while loop is only executed
# if the break statement is not executed
# if the break statement is executed
# else block is not executed

# for else
for x in my_list:
    if x == 'C#':
        result = 'Found C#'
        break
else:
    result = 'C# not found'

print(result)
# Prints 'C# not found'
# same as while...else
# else is only executed if break is not executed

Beginner Python tips Day - 04 random

Paurakh Sharma Humagain — Sun, 07 Jun 2020 13:14:34 +0000

# Python Day - 04 random

import random
my_list = ['Python', 'JavaScript', 'Dart', 'Golang', 'Rust']

# choice - get a random element from a list
print(random.choice(my_list))
# Prints a single element from the list e.g 'Dart'

# choices - get number of elements from a list with replacement
print(random.choices(my_list, k=2))
# Prints a list of two elements from a list e.g ['Rust', 'Golang']
# Note: choices returns the elements with replacement meaning,
# You might get the same element more than once e.g ['Rust', 'Rust']

# sample - get number of elements from a list without replacement
print(random.sample(my_list, k=2))
# Similar to choices but you will not get same element more than once

# shuffle - shuffle the elements in a list
random.shuffle(my_list)
print(my_list)
# Prints e.g ['Golang', 'Dart', 'JavaScript', 'Python', 'Rust']
# Randomize the original list

Beginner Python tips Day - 03 defaultdict

Paurakh Sharma Humagain — Sat, 06 Jun 2020 08:06:15 +0000

# Python Day-03 defaultdict

from collections import defaultdict

# You can specify the type of values the dictionary will contain
dict_of_list = defaultdict(list)

dict_of_list['languages'] = ['Python', 'JavaScript', 'Dart']

print(dict_of_list['languages'])
# Prints ['Python', 'JavaScript', 'Dart']

# when the key cannot be found empty list is returned
print(dict_of_list['frameworks'])
# Prints [] i.e empty list

dict_of_numbers = defaultdict(int)

dict_of_numbers['students'] = 10

print(dict_of_numbers['students'])
# Prints 10

print(dict_of_numbers['teachers'])
# Prints 0

# If the key isn't found return the predefined value
safe_dict = defaultdict(lambda : None)

safe_dict['hello'] = 'world'

print(safe_dict['hello'])
# Prints 'world'

print(safe_dict['world'])
# Prints None

Beginner Python tips Day - 02 dict.get()

Paurakh Sharma Humagain — Fri, 05 Jun 2020 08:27:26 +0000

# Day-02 dict .get()

my_dict = {'hello': 'world'}

try:
    my_dict['world']
    # Throws KeyError exception
except KeyError:
    print('Cannot find the provided key in the dictionary')

# Use .get() to be safe from the exception
print(my_dict.get('world'))
# prints None in the console

# You can pass second argument as default value
print(my_dict.get('world', 'default value'))
# prints 'default value'

Beginner Python tips Day - 01 Enumerate

Paurakh Sharma Humagain — Thu, 04 Jun 2020 03:49:47 +0000

# Day 01 - Enumerate
# Provides counter that matches
# the number of items in the list

my_list = ['hello', 'world', '!']

# Without enumerate
for index in range(len(my_list)):
    print(index, my_list[index])
'''
Prints
0 hello
1 world
2 !
'''

# With enumerate
for counter, value in enumerate(my_list):
    print(counter, value)
'''
Prints
0 hello
1 world
2 !
'''

# Don't confuse it with index
# because, the counter can start from any integer
for counter, value in enumerate(my_list, 2):
    print(counter, value)
'''
Prints
2 hello
3 world
4 !
'''

Microservice in Python using FastAPI

Paurakh Sharma Humagain — Thu, 19 Mar 2020 11:43:00 +0000

Creating Microservices with Python

Introduction to Microservices
- Benefits of Microservice
- Drawbacks of Microservice
- Why Microservice in Python
Introduction to FastAPI
- Why FastAPI
- Installing FastAPI
- Creating Simple REST API using FastAPI
- Using PostgreSQL Database with FastAPI
Microservice Data Management Patterns
- Database Per Service
- Shared Database
- API Composition
Creating a Python Microservice in Docker
- Installing Docker and Docker Compose
- Creating Movies Service
- Creating Casts Service
- Running the microservice using Docker Compose
- Using Nginx to Access Both Services using a Single Host Address
Conclusion and Next Step

As a Python developer you might have heard about the term microservices, and want to build a Python microservice by yourself. Microservices is a great architecture for building highly scalable applications. Before starting to build your application using the microservice, you must be familiar with the benefits and drawbacks of using microservices. In this article, you will learn the benefits and drawbacks of using microservices. You will also learn how you can build your own microservice and deploy it using the Docker Compose.

In this tutorial you'll learn:

What the benefits and drawbacks of microservices are
Why you should build microservice with Python
How to build REST API using FastAPI and PostgreSQL
How to build microservice using FastAPI
How to run microservices using docker-compose
How to manage microservices using Nginx

You will first build a simple REST API using FastAPI and then use PostgreSQL as our database. You will then extend the same application to a microservice.

Introduction to Microservices

Microservice is the approach of breaking down large monolith application into individual applications specializing in a specific service/functionality. This approach is often known as Service-Oriented Architecture or SOA.

In monolithic architecture, every business logic resides in the same application. Application services such as user management, authentication, and other features use the same database.

In a microservice architecture, the application is broken down into several separate services that run in separate processes. There is a different database for different functionality of the application and the services communicate with each other using the HTTP, AMQP, or a binary protocol like TCP, depending on the nature of each service. Inter-service communication can also be performed using the message queues like RabbitMQ, Kafka or Redis.

Benefits of Microservice

The microservice architecture comes with lots of benefits. Some of these benefits are:

Loosely coupled application means the different services can be build using the technologies that suit them best. So, the development team is not bounded to the choices made while starting the project.
Since the services are responsible for specific functionality which makes it easier to understand and keep the application under control.
Application scaling also becomes easier because if one of the services requires high GPU usage then only the server consisting that service needs to have high GPU and others can run on a normal server.

Drawbacks of Microservice

The microservice architecture is not a silver bullet that solves all your problems, it comes with its drawbacks too. Some of these drawbacks are:

Since different services use the different database the transactions involving more than one service needs to use eventual consistency.
Perfect splitting of the services is very difficult to achieve at the first try and this needs to be iterated before coming with the best possible separation of the services.
Since the services communicate with each other through the use of network interaction, this makes the application slower due to the network latency and slow service.

Why Microservice in Python

Python is a perfect tool for building micro-services because it comes with a great community, easy learning curve and tons of libraries. Due to the introduction of asynchronous programming in Python, web frameworks with performance on-par with GO and Node.js, has emerged.

Introduction to FastAPI

FastAPI is a modern, high-performance, web framework, which comes with tons of cool features like auto-documentation based on OpenAPI and built-in serialization and validation library. See here for the list of all cool features in FastAPI.

Why FastAPI

Some of the reason why I think FastAPI is a great choice for building microservices in Python is:

Auto documentation
Async/Await support
Built-in validation and serialization
100% type annotated so autocompletion works great

Installing FastAPI

Before installing FastAPI create a new directory movie_service and create a new virtual environment inside the newly created directory using virtualenv.
If you haven't already installed virtualenv:

pip install virtualenv

Now, create a new virtual environment.

virtualenv env

If you are on Mac/Linux you can activate the virtual environment using the command:

source ./env/bin/activate

Windows users can run this command instead:

.\env\Scripts\activate

Finally, Your are ready to install FastAPI, run the following command:

pip install fastapi

Since FastAPI doesn't come with inbuilt service, you need to install uvicorn for it to run. uvicorn is an ASGI server which allows us to use async/await features.
Install uvicorn using the command

pip install uvicorn

Creating Simple REST API using FastAPI

Before You start building a microservice using FastAPI, let's learn the basics of FastAPI. Create a new directory app and a new file main.py inside the newly-created directory.

Add the following code in main.py.

#~/movie_service/app/main.py

from fastapi import FastAPI

app = FastAPI()


@app.get('/')
async def index():
    return {"Real": "Python"}

Here you first import and instantiate the FastAPI and then register the root endpoint / which then returns a JSON.

You can run the application server using uvicorn app.main:app --reload. Here app.main indicates you use main.py file inside the app directory and :app indicates our FastAPI instance name.

You can access the app from http://127.0.0.1:8000. To access the cool automatic documentation, head over to http://127.0.0.1:8000/docs. You can play around and interact with your API from the browser itself.

Let's add some CRUD functionality to our application.
Update your main.py to look like the following:

#~/movie_service/app/main.py

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

app = FastAPI()

fake_movie_db = [
    {
        'name': 'Star Wars: Episode IX - The Rise of Skywalker',
        'plot': 'The surviving members of the resistance face the First Order once again.',
        'genres': ['Action', 'Adventure', 'Fantasy'],
        'casts': ['Daisy Ridley', 'Adam Driver']
    }
]

class Movie(BaseModel):
    name: str
    plot: str
    genres: List[str]
    casts: List[str]


@app.get('/', response_model=List[Movie])
async def index():
    return fake_movie_db

As you can see you have created a new class Movie which extends BaseModel from pydantic.
The Movie model contains the name, phot, genres, and casts. Pydantic comes built-in with FastAPI which makes making models and request validation a breeze.

If you head over to the docs site you can see there are fields of our Movies model mentioned already in the example response section. This is possible because you have defined the response_model in our route definition.

Now, let's add the endpoint to add a movie to our movies list.

Add a new endpoint definition to handle the POST request.

@app.post('/', status_code=201)
async def add_movie(payload: Movie):
    movie = payload.dict()
    fake_movie_db.append(movie)
    return {'id': len(fake_movie_db) - 1}

Now, head over to the browser and test the new API. Try adding a movie with an invalid field or without the required fields and see that the validation is automatically handled by FastAPI.

Let's add a new endpoint to update the movie.

@app.put('/{id}')
async def update_movie(id: int, payload: Movie):
    movie = payload.dict()
    movies_length = len(fake_movie_db)
    if 0 <= id <= movies_length:
        fake_movie_db[id] = movie
        return None
    raise HTTPException(status_code=404, detail="Movie with given id not found")

Here id is the index of our fake_movie_db list.

Note: Remember to import HTTPException from fastapi

Now you can also add the endpoint to delete the movie.

@app.delete('/{id}')
async def delete_movie(id: int):
    movies_length = len(fake_movie_db)
    if 0 <= id <= movies_length:
        del fake_movie_db[id]
        return None
    raise HTTPException(status_code=404, detail="Movie with given id not found")

Before you move forward, let's structure our app in a better way. Create a new folder api inside app and create a new file movies.py inside the recently created folder. Move all the routes related codes from main.py to movies.py. So, the movies.py should look like the following:

#~/movie-service/app/api/movies.py

from typing import List
from fastapi import Header, APIRouter

from app.api.models import Movie

fake_movie_db = [
    {
        'name': 'Star Wars: Episode IX - The Rise of Skywalker',
        'plot': 'The surviving members of the resistance face the First Order once again.',
        'genres': ['Action', 'Adventure', 'Fantasy'],
        'casts': ['Daisy Ridley', 'Adam Driver']
    }
]

movies = APIRouter()

@movies.get('/', response_model=List[Movie])
async def index():
    return fake_movie_db

@movies.post('/', status_code=201)
async def add_movie(payload: Movie):
    movie = payload.dict()
    fake_movie_db.append(movie)
    return {'id': len(fake_movie_db) - 1}

@movies.put('/{id}')
async def update_movie(id: int, payload: Movie):
    movie = payload.dict()
    movies_length = len(fake_movie_db)
    if 0 <= id <= movies_length:
        fake_movie_db[id] = movie
        return None
    raise HTTPException(status_code=404, detail="Movie with given id not found")

@movies.delete('/{id}')
async def delete_movie(id: int):
    movies_length = len(fake_movie_db)
    if 0 <= id <= movies_length:
        del fake_movie_db[id]
        return None
    raise HTTPException(status_code=404, detail="Movie with given id not found")

Here you registered a new API route using the APIRouter from FastAPI.

Also, create a new file models.py inside api where you will be keeping our Pydantic models.

#~/movie-service/api/models.py

from typing import List
from pydantic import BaseModel

class Movie(BaseModel):
    name: str
    plot: str
    genres: List[str]
    casts: List[str]

Now register this new routes file in main.py

#~/movie-service/app/main.py
from fastapi import FastAPI

from app.api.movies import movies

app = FastAPI()

app.include_router(movies)

Finally, our application directory structure looks like this:

movie-service
├── app
│   ├── api
│   │   ├── models.py
│   │   ├── movies.py
│   |── main.py
└── env

Make sure your application is working properly before you move forward.

Using PostgreSQL Database with FastAPI

Previously, you used fake Python list to add movies but now you are finally ready to use an actual database for this purpose. You are going to use PostgreSQL for this purpose. Install PostgreSQL if you haven't already. After installing the PostgreSQl create a new database, I am going to call mine movie_db.

You are going to use encode/databases to connect to the database using async and await support. Learn more about async/await in Python here

Install the required library using:

pip install 'databases[postgresql]'

this will install sqlalchemy and asyncpg as well, which are required for working with PostgreSQL.

Create a new file inside api and call it db.py. This file will contain the actual database model for our REST API.

#~/movie-service/app/api/db.py

from sqlalchemy import (Column, Integer, MetaData, String, Table,
                        create_engine, ARRAY)

from databases import Database

DATABASE_URL = 'postgresql://movie_user:movie_password@localhost/movie_db'

engine = create_engine(DATABASE_URL)
metadata = MetaData()

movies = Table(
    'movies',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('plot', String(250)),
    Column('genres', ARRAY(String)),
    Column('casts', ARRAY(String))
)

database = Database(DATABASE_URL)

Here, DATABASE_URI is the URL used to connect to the PostgreSQL database. Here movie_user is the name of the database user, movie_password is the password of the database user and movie_db is the name of the database.

Just like you would to in SQLAlchemy you have created the table for the movies database.

Update main.py to connect to the database. main.py should look like the following:

#~/movie-service/app/main.py

from fastapi import FastAPI
from app.api.movies import movies
from app.api.db import metadata, database, engine

metadata.create_all(engine)

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


app.include_router(movies)

FastAPI provides some event handlers which you can use to connect to our database when the application starts and disconnect when it shuts down.

Update movies.py so that it uses a database instead of a fake Python list.

#~/movie-service/app/api/movies.py


from typing import List
from fastapi import Header, APIRouter

from app.api.models import MovieIn, MovieOut
from app.api import db_manager

movies = APIRouter()

@movies.get('/', response_model=List[MovieOut])
async def index():
    return await db_manager.get_all_movies()

@movies.post('/', status_code=201)
async def add_movie(payload: MovieIn):
    movie_id = await db_manager.add_movie(payload)
    response = {
        'id': movie_id,
        **payload.dict()
    }

    return response

@movies.put('/{id}')
async def update_movie(id: int, payload: MovieIn):
    movie = payload.dict()
    fake_movie_db[id] = movie
    return None

@movies.put('/{id}')
async def update_movie(id: int, payload: MovieIn):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")

    update_data = payload.dict(exclude_unset=True)
    movie_in_db = MovieIn(**movie)

    updated_movie = movie_in_db.copy(update=update_data)

    return await db_manager.update_movie(id, updated_movie)

@movies.delete('/{id}')
async def delete_movie(id: int):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")
    return await db_manager.delete_movie(id)

Let's add db_manager.py to manipulate our database.

#~/movie-service/app/api/db_manager.py

from app.api.models import MovieIn, MovieOut, MovieUpdate
from app.api.db import movies, database


async def add_movie(payload: MovieIn):
    query = movies.insert().values(**payload.dict())

    return await database.execute(query=query)

async def get_all_movies():
    query = movies.select()
    return await database.fetch_all(query=query)

async def get_movie(id):
    query = movies.select(movies.c.id==id)
    return await database.fetch_one(query=query)

async def delete_movie(id: int):
    query = movies.delete().where(movies.c.id==id)
    return await database.execute(query=query)

async def update_movie(id: int, payload: MovieIn):
    query = (
        movies
        .update()
        .where(movies.c.id == id)
        .values(**payload.dict())
    )
    return await database.execute(query=query)

Let's update our models.py so that you can use the Pydantic model with the sqlalchemy table.

#~/movie-service/app/api/models.py

from pydantic import BaseModel
from typing import List, Optional

class MovieIn(BaseModel):
    name: str
    plot: str
    genres: List[str]
    casts: List[str]


class MovieOut(MovieIn):
    id: int


class MovieUpdate(MovieIn):
    name: Optional[str] = None
    plot: Optional[str] = None
    genres: Optional[List[str]] = None
    casts: Optional[List[str]] = None

Here MovieIn is the base model that you use to add the movie to the database. You have to add the id to this model while getting it from the database, hence the MovieOut model. MovieUpdate model allows us to set the values in the model to be optional so that while updating the movie only the field that needs to be updated can be sent.

Now, head over to the browser documentation site and start playing with the API.

Microservice Data Management Patterns

Managing data in microservice is one of the most challenging aspects of building a microservice. Since different functions of the application are handled by different services, usage of a database can be tricky.

Here are some patterns that you can use to manage data flow in the application.

Database Per Service

Using a database per service is great if you want your microservices to be as loosely coupled as possible. Having a different database per service allows us to scale different services independently. A transaction involving multiple databases is done through well-defined APIs. This comes with its drawback as implementing business transactions involving multiple services is not straightforward. Also, the addition of network overhead makes this less efficient to use.

Shared Database

If there are lots of transactions involving multiple services it is better to use a shared database. This comes with the benefits of highly consistent application but takes away most of the benefits that come with the microservices architecture. Developers working on one service needs to coordinate with the schema changes in other services.

API Composition

In transactions involving multiple databases, API composer acts as an API gateway and executes API calls to other microservices in the required order. Finally, the results from each microservices are returned to the client service after performing an in-memory join. The downside of this approach is inefficient in-memory joins of a large dataset.

Creating a Python Microservice in Docker

The pain of deploying the microservice can be greatly reduced by using the Docker. Docker helps to encapsulate each service and scale them independently.

Installing Docker and Docker Compose

If you haven't already install docker in your system. Verify if the docker is installed by running the command docker. After you have done installing Docker, install Docker Compose. Docker Compose is used for defining and running multiple Docker containers. It also helps in easy interaction between them.

Creating Movies Service

Since a lot of the work for building a movie service is already done while getting started with the FastAPI, you are going to reuse the code you have already written. Create a brand new folder, I am going to call mine python-microservices. Move the code you wrote earlier which I had named movie-service.
So, the folder structure would look like this:

python-microservices/
└── movie-service/
    ├── app/
    └── env/

First of all, let's create a requirements.txt file where you are going to keep all the dependencies you are going to use in our movie-service.
Create a new file requirements.txt inside movie-service and add the following to it:

asyncpg==0.20.1
databases[postgresql]==0.2.6
fastapi==0.48.0
SQLAlchemy==1.3.13
uvicorn==0.11.2
httpx==0.11.1

You have used all the libraries mentioned there except httpx which you are going to use while making service to service API call.

Create a Dockerfile inside movie-service with the following contents:

FROM python:3.8-slim

WORKDIR /app

COPY ./requirements.txt /app/requirements.txt

RUN apt-get update \
    && apt-get install gcc -y \
    && apt-get clean

RUN pip install -r /app/requirements.txt \
    && rm -rf /root/.cache/pip

COPY . /app/

Here first, you define which Python version you want to use. Then set the WORKDIR to be inside app folder inside the Docker container. After that gcc is installed which is required by the libraries that you are using in the application.
Finally, install all dependencies in requirements.txt and copy all the files inside movie-service/app.

Update db.py and replace

DATABASE_URI = 'postgresql://movie_user:movie_password@localhost/movie_db'

with

DATABASE_URI = os.getenv('DATABASE_URI')

Note: Don't forget to import os on the top of the file.

You need to do this so that you can latter provide DATABASE_URI as an environment variable.

Also, update main.py and replace

app.include_router(movies)

with

app.include_router(movies, prefix='/api/v1/movies', tags=['movies'])

Here, you have added prefix /api/v1/movies so, that managing different version of API becomes easier. Also, tags make finding API related to movies easier in FastAPI docs.

Also, you need to update our models so that the casts stores the cast's id instead of the actual name. So, update the models.py to look like this:

#~/python-microservices/movie-service/app/api/models.py

from pydantic import BaseModel
from typing import List, Optional

class MovieIn(BaseModel):
    name: str
    plot: str
    genres: List[str]
    casts_id: List[int]


class MovieOut(MovieIn):
    id: int


class MovieUpdate(MovieIn):
    name: Optional[str] = None
    plot: Optional[str] = None
    genres: Optional[List[str]] = None
    casts_id: Optional[List[int]] = None

Likewise, you need to update the database tables, let's update db.py:

#~/python-microservices/movie-service/app/api/db.py

import os

from sqlalchemy import (Column, DateTime, Integer, MetaData, String, Table,
                        create_engine, ARRAY)

from databases import Database

DATABASE_URL = os.getenv('DATABASE_URL')

engine = create_engine(DATABASE_URL)
metadata = MetaData()

movies = Table(
    'movies',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('plot', String(250)),
    Column('genres', ARRAY(String)),
    Column('casts_id', ARRAY(Integer))
)

database = Database(DATABASE_URL)

Now, update movies.py to check if the cast with the given id presents in cast service before adding a new movie or updating a movie.

#~/python-microservices/movie-service/app/api/movies.py

from typing import List
from fastapi import APIRouter, HTTPException

from app.api.models import MovieOut, MovieIn, MovieUpdate
from app.api import db_manager
from app.api.service import is_cast_present

movies = APIRouter()

@movies.post('/', response_model=MovieOut, status_code=201)
async def create_movie(payload: MovieIn):
    for cast_id in payload.casts_id:
        if not is_cast_present(cast_id):
            raise HTTPException(status_code=404, detail=f"Cast with id:{cast_id} not found")

    movie_id = await db_manager.add_movie(payload)
    response = {
        'id': movie_id,
        **payload.dict()
    }

    return response

@movies.get('/', response_model=List[MovieOut])
async def get_movies():
    return await db_manager.get_all_movies()

@movies.get('/{id}/', response_model=MovieOut)
async def get_movie(id: int):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")
    return movie

@movies.put('/{id}/', response_model=MovieOut)
async def update_movie(id: int, payload: MovieUpdate):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")

    update_data = payload.dict(exclude_unset=True)

    if 'casts_id' in update_data:
        for cast_id in payload.casts_id:
            if not is_cast_present(cast_id):
                raise HTTPException(status_code=404, detail=f"Cast with given id:{cast_id} not found")

    movie_in_db = MovieIn(**movie)

    updated_movie = movie_in_db.copy(update=update_data)

    return await db_manager.update_movie(id, updated_movie)

@movies.delete('/{id}', response_model=None)
async def delete_movie(id: int):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")
    return await db_manager.delete_movie(id)

Let's add a service to make an API call to cast service:

#~/python-microservices/movie-service/app/api/service.py

import os
import httpx

CAST_SERVICE_HOST_URL = 'http://localhost:8002/api/v1/casts/'
url = os.environ.get('CAST_SERVICE_HOST_URL') or CAST_SERVICE_HOST_URL

def is_cast_present(cast_id: int):
    r = httpx.get(f'{url}{cast_id}')
    return True if r.status_code == 200 else False

You make an api call to get the cast with the given id and return true if the cast exists and false otherwise.

Creating Casts Service

Similar to a movie-service, for creating a casts-service you are going to use FastAPI and PostgreSQL database.

Create a folder structure like the following:

python-microservices/
.
├── cast_service/
│   ├── app/
│   │   ├── api/
│   │   │   ├── casts.py
│   │   │   ├── db_manager.py
│   │   │   ├── db.py
│   │   │   ├── models.py
│   │   ├── main.py
│   ├── Dockerfile
│   └── requirements.txt
├── movie_service/
...

Add the following to requirements.txt:

asyncpg==0.20.1
databases[postgresql]==0.2.6
fastapi==0.48.0
SQLAlchemy==1.3.13
uvicorn==0.11.2

Dockerfile:

FROM python:3.8-slim

WORKDIR /app

COPY ./requirements.txt /app/requirements.txt

RUN apt-get update \
    && apt-get install gcc -y \
    && apt-get clean

RUN pip install -r /app/requirements.txt \
    && rm -rf /root/.cache/pip

COPY . /app/

main.py

#~/python-microservices/cast-service/app/main.py

from fastapi import FastAPI
from app.api.casts import casts
from app.api.db import metadata, database, engine

metadata.create_all(engine)

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

app.include_router(casts, prefix='/api/v1/casts', tags=['casts'])

You have added the prefix of /api/v1/casts so that managing the API becomes easier. Also, adding tags makes finding docs related to casts in the FastAPI docs easier.

casts.py

#~/python-microservices/cast-service/app/api/casts.py

from fastapi import APIRouter, HTTPException
from typing import List

from app.api.models import CastOut, CastIn, CastUpdate
from app.api import db_manager

casts = APIRouter()

@casts.post('/', response_model=CastOut, status_code=201)
async def create_cast(payload: CastIn):
    cast_id = await db_manager.add_cast(payload)

    response = {
        'id': cast_id,
        **payload.dict()
    }

    return response

@casts.get('/{id}/', response_model=CastOut)
async def get_cast(id: int):
    cast = await db_manager.get_cast(id)
    if not cast:
        raise HTTPException(status_code=404, detail="Cast not found")
    return cast

db_manager.py

#~/python-microservices/cast-service/app/api/db_manager.py

from app.api.models import CastIn, CastOut, CastUpdate
from app.api.db import casts, database


async def add_cast(payload: CastIn):
    query = casts.insert().values(**payload.dict())

    return await database.execute(query=query)

async def get_cast(id):
    query = casts.select(casts.c.id==id)
    return await database.fetch_one(query=query)

db.py

#~/python-microservices/cast-service/app/api/db.py

import os

from sqlalchemy import (Column, Integer, MetaData, String, Table,
                        create_engine, ARRAY)

from databases import Database

DATABASE_URI = os.getenv('DATABASE_URI')

engine = create_engine(DATABASE_URI)
metadata = MetaData()

casts = Table(
    'casts',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('nationality', String(20)),
)

database = Database(DATABASE_URI)

models.py

#~/python-microservices/cast-service/app/api/models.py

from pydantic import BaseModel
from typing import List, Optional

class CastIn(BaseModel):
    name: str
    nationality: Optional[str] = None


class CastOut(CastIn):
    id: int


class CastUpdate(CastIn):
    name: Optional[str] = None

Running the microservice using Docker Compose

To run the microservices, create a docker-compose.yml file and add the following to it:

version: '3.7'

services:
  movie_service:
    build: ./movie-service
    command: uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
    volumes:
      - ./movie-service/:/app/
    ports:
      - 8001:8000
    environment:
      - DATABASE_URI=postgresql://movie_db_username:movie_db_password@movie_db/movie_db_dev
      - CAST_SERVICE_HOST_URL=http://cast_service:8000/api/v1/casts/

  movie_db:
    image: postgres:12.1-alpine
    volumes:
      - postgres_data_movie:/var/lib/postgresql/data/
    environment:
      - POSTGRES_USER=movie_db_username
      - POSTGRES_PASSWORD=movie_db_password
      - POSTGRES_DB=movie_db_dev

  cast_service:
    build: ./cast-service
    command: uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
    volumes:
      - ./cast-service/:/app/
    ports:
      - 8002:8000
    environment:
      - DATABASE_URI=postgresql://cast_db_username:cast_db_password@cast_db/cast_db_dev

  cast_db:
    image: postgres:12.1-alpine
    volumes:
      - postgres_data_cast:/var/lib/postgresql/data/
    environment:
      - POSTGRES_USER=cast_db_username
      - POSTGRES_PASSWORD=cast_db_password
      - POSTGRES_DB=cast_db_dev

volumes:
  postgres_data_movie:
  postgres_data_cast:

Here you have 4 different services, movie_service, a database for movie_service, cast_service and a database for cast service. You have exposed movie_service to port 8001 similarly cast_service to port 8002.

For the database, you have used volumes so that the data doesn't get destroyed when the docker container is shutdown.

Run the docker-compose using the command:

docker-compose up -d

This creates the docker image if it doesn't already exist and runs them.

Head over to http://localhost:8002/docs to add a cast in casts service. Similarly, http://localhost:8001/docs to add the movie in the movie service.

Using Nginx to Access Both Services using a Single Host Address

You have deployed the microservices using Docker compose but there is one minor problem. Each of the microservices needs to be accessed using a different Port. You can solve this issue using Nginx reverse proxy, using Nginx you can direct the request add a middleware which routes our requests to different services based on the API URL.

Add a new file nginx_config.conf inside python-microservices with the following contents.


server {
  listen 8080;

  location /api/v1/movies {
    proxy_pass http://movie_service:8000/api/v1/movies;
  }

  location /api/v1/casts {
    proxy_pass http://cast_service:8000/api/v1/casts;
  }

}

Here you are running the Nginx at port 8080 and routing the requests to movie service if the endpoint starts with /api/v1/movies and similarly to cast service if the endpoint starts with /api/v1/casts

Now, you need to add the nginx service in our docker-compose-yml. Add the following service after cast_db service:

nginx:
    image: nginx:latest
    ports:
      - "8080:8080"
    volumes:
      - ./nginx_config.conf:/etc/nginx/conf.d/default.conf
    depends_on:
      - cast_service
      - movie_service

Now, shut down the containers with the command:

docker-compose down

And run it back again with:

docker-compose up -d

Now, you can access both movie service and cast service at port 8080.
Head over to http://localhost:8080/api/v1/movies/ to get the list of movies.

Now, you might be wondering how you can access the docs of the services. For that update main.py of movie service and replace

app = FastAPI()

with

app = FastAPI(openapi_url="/api/v1/movies/openapi.json", docs_url="/api/v1/movies/docs")

Similarly, for cast service replace it with

app = FastAPI(openapi_url="/api/v1/casts/openapi.json", docs_url="/api/v1/casts/docs")

Here, you changed which endpoint the docs are served and from where the openapi.json is served.

Now, you can access the docs from http://localhost:8080/api/v1/movies/docs for movie service and from http://localhost:8080/api/v1/casts/docs for casts service.

If you are stuck at some point or just want to have a look at the complete code, head over to the Github Repo

Conclusion and Next Step

The microservice architecture is great for breaking down a large monolith application into separate business logics but this comes with the complication too. Python is great for building microservice because of the developer experience and tons of packages and frameworks to make developers more productive.

Deploying microservices has become easier thanks to Docker. Learn more on How to develop microservices using Docker, and Docker Compose

Want me to cover any topic? Let me know at twitter or write a comment down below.

Flask Rest API -Part:6- Testing REST APIs

Paurakh Sharma Humagain — Mon, 03 Feb 2020 05:01:07 +0000

Part 6: Testing REST APIs

Howdy! In the previous Part of the series, we learned how to perform
password reset in our REST API.

In this part, we are going to learn how to test our REST API endpoints.

Why should we spend time writing tests?

To make sure our application doesn't break while making changes/refactoring
To automate repetitive manual tests reducing human errors
Be able to release to the production on Fridays ;)
Testing provides a better CI/ CD workflow.

I hope you are convinced that we should write tests. Let's get started with testing our Flask application.

When it comes to testing, there are two most popular tools to test Python applications.
1) unittest: unittest is a python standard library which means it is distributed with Python. unittest provides tons of tools for constructing and running tests.

2) pytest: pytest is a python library which I like to call is the superset of unittest which means you can run tests written in unittest with pytest. It makes writing tests easier and faster.

In this tutorial, we are going to learn how to write tests using unittest, because it enables us to write our tests using OOP.

Before we start, remove the line below from app.py

app.config['MONGODB_SETTINGS'] = {
    'host': 'mongodb://localhost/movie-bag'
}

and add

MONGODB_SETTINGS = {
    'host': 'mongodb://localhost/movie-bag'
}

to our .env, this step is required because we want to use a different database for developing our application and running the tests.

First of all, let's create an env file to store our test-related configurations, we should separate our test configs from our development and production configs.

In the root directory create a file .env.test and add the following configs to it.

touch .env.test

#~/movie-bag/.env.test

JWT_SECRET_KEY = 'super-secret'
MAIL_SERVER: "localhost"
MAIL_PORT = "1025"
MAIL_USERNAME = "support@movie-bag.com"
MAIL_PASSWORD = ""
MONGODB_SETTINGS = {
    'host': 'mongodb://localhost/movie-bag-test'
}

Notice that we have used different database for our test config, this is done because our tests and we want our tests and development database to be separated. We also want our test database to be empty before running the tests.

Now, let's create a new folder tests inside our root directory. Create a new file __init__.py inside the tests folder, also, create a new file test_signup.py.

mkdir tests
cd tests
touch __init__.py
touch test_signup.py

#~/movie-bag/tests/test_signup.py

import unittest
import json

from app import app
from database.db import db


class SignupTest(unittest.TestCase):

    def setUp(self):
        self.app = app.test_client()
        self.db = db.get_db()

    def test_successful_signup(self):
        # Given
        payload = json.dumps({
            "email": "paurakh011@gmail.com",
            "password": "mycoolpassword"
        })

        # When
        response = self.app.post('/api/auth/signup', headers={"Content-Type": "application/json"}, data=payload)

        # Then
        self.assertEqual(str, type(response.json['id']))
        self.assertEqual(200, response.status_code)

    def tearDown(self):
        # Delete Database collections after the test is complete
        for collection in self.db.list_collection_names():
            self.db.drop_collection(collection)

Let's go step by step to understand what is actually going on.

First of all, we define SignupTest class which extends unittest.TestCase. TestCase provides us with useful methods such as setUp and tearDown and also the assertation methods.

setUp() method runs each time before running each method defined on the SignupTest class. setUp() as the name suggests is used to set up our test infrastructure before running the tests.

Here you can see we define this.app and this.db in this method. We use app.test_client() instead of app because it makes testing our flask application easier. Also, we get our Database instance with db.get_db() and set it to this.db.

Similarly, test_successful_signup() is the method that is actually testing the Signup feature. Here we have defined a payload which should be a JSON value. And we send a POST request to /api/auth/signup.

The response from the request is used to finally assert that our Signup feature actually sent the user id and successful status code which is 200.

Finally, after each test methods the tearDown() method runs each time. This method is responsible for clearing our infrastructure setup. This includes deleting our database collection for test isolation.

Test Isolation

Test isolation is one of the most important concepts in testing. Usually, when we are writing tests, we test one business logic. The idea of test isolation is that one of your tests should not in any way affect another test.

Suppose that you have created a user in one test and you are testing login on another test. To follow test isolation you cannot depend on the user-created in a user creation test, but should create the user right in the test where you are going to test login. Why? Because your login test might run before your user creation test this makes your test fail.

Also, if we do not delete our user which we created on the previous test run, and we try to run the test again, our test fails because the user is already there.
So, we should always test a feature from an empty state and for that easiest way is to delete all the collections in our database.

Before running our first test make sure to export environment variable ENV_FILE_LOCATION with the location to the test env file.

To set this value mac/linux can run the command:

export ENV_FILE_LOCATION=./.env.test

and windows user can run the command:

set ENV_FILE_LOCATION=./.env.test

Make sure you have activated your virtual environment with pipenv shell.

To run the test enter this command in your terminal.

python -m unittest tests/test_signup.py

You should be able to see the output like this:

.
----------------------------------------------------------------------
Ran 1 test in 1.023s

OK

This means our test run successfully.

If you run into any error feel free to comment down, I am always ready to help you out

As you can see we are going to need this setUp() and tearDown() in our ever TestCase. So, let's move this logic to a new file, let's call it BaseCase.py.

#~/movie-bag/tests/BaseCase.py

import unittest

from app import app
from database.db import db


class BaseCase(unittest.TestCase):

    def setUp(self):
        self.app = app.test_client()
        self.db = db.get_db()


    def tearDown(self):
        # Delete Database collections after the test is complete
        for collection in self.db.list_collection_names():
            self.db.drop_collection(collection)

Now update your test_signup.py to look like this:


 import json

-from app import app
-from database.db import db
+from tests.BaseCase import BaseCase

-
-class SignupTest(unittest.TestCase):
+class SignupTest(BaseCase):
-
-    def setUp(self):
-        self.app = app.test_client()
-        self.db = db.get_db()

     def test_successful_signup(self):
         # Given
...
-
-    def tearDown(self):
-        # Delete Database collections after the test is complete
-        for collection in self.db.list_collection_names():
-            self.db.drop_collection(collection)

Now let's add test for our Login feature, create a new file test_login.py inside tests folder with the following code.

#~/movie-bag/tests/test_login.py

import json

from tests.BaseCase import BaseCase

class TestUserLogin(BaseCase):

    def test_successful_login(self):
        # Given
        email = "paurakh011@gmail.com"
        password = "mycoolpassword"
        payload = json.dumps({
            "email": email,
            "password": password
        })
        response = self.app.post('/api/auth/signup', headers={"Content-Type": "application/json"}, data=payload)

        # When
        response = self.app.post('/api/auth/login', headers={"Content-Type": "application/json"}, data=payload)

        # Then
        self.assertEqual(str, type(response.json['token']))
        self.assertEqual(200, response.status_code)

Here we first created the user with /api/auth/signup endpoint and login using the same email and password and assert that the /api/auth/login endpoint returns the token.

Now, let's add tests to check the creation of the movie.
Create test_create_movie.py with the code below.

#movie-bag/tests/test_create_movie.py

import json

from tests.BaseCase import BaseCase

class TestUserLogin(BaseCase):

    def test_successful_login(self):
        # Given
        email = "paurakh011@gmail.com"
        password = "mycoolpassword"
        user_payload = json.dumps({
            "email": email,
            "password": password
        })

        self.app.post('/api/auth/signup', headers={"Content-Type": "application/json"}, data=user_payload)
        response = self.app.post('/api/auth/login', headers={"Content-Type": "application/json"}, data=user_payload)
        login_token = response.json['token']

        movie_payload = {
            "name": "Star Wars: The Rise of Skywalker",
            "casts": ["Daisy Ridley", "Adam Driver"],
            "genres": ["Fantasy", "Sci-fi"]
        }
        # When
        response = self.app.post('/api/movies',
            headers={"Content-Type": "application/json", "Authorization": f"Bearer {login_token}"},
            data=json.dumps(movie_payload))

        # Then
        self.assertEqual(str, type(response.json['id']))
        self.assertEqual(200, response.status_code)

To run all the tests at once use the command:

python -m unittest --buffer

Here --buffer or -b is used to discard the output on a successful test run.

Here we first signup for the user account, log in as the user to get the login token and then use the login token to create a movie. Finally, we check to see if the movie creating endpoint returns the id to the created movie.

You might have noticed in this test we only check if the movie creation works but do not check if the user creation worked or user login worked. This is because we already have separate tests that are testing these things so, we don't have to repeat the same tests.

We have only created happy path tests but it is crucial for us to test that our application response is expected even in the case when the user enters invalid input. For instance, the user doesn't send the password while signing up or sends an invalid format email.

I have not included these tests in the tutorial itself, but I will be sure to include them in the Github repo.

You can find all the code we have written till now and more tests here

What we learned from this part of the series?

Why we should write tests for our application
What test isolation is and why we should isolate our tests cases
How to test Flask REST APIs with unittest

Feel free to add anything I am missing in this article on the comments below.

If you have any topic suggestions, please let me know. I hope to see you in the next one.

Until then, you can follow me on twitter

Forem: Paurakh Sharma Humagain

Using Firebase Cloud Functions to Generate URL Preview

Using Firebase Cloud Functions to Generate URL Preview

How this works?

Let's get started

Floating Audio Player in Flutter

Floating Audio Player in Flutter

Extend your Flutter code

Extend your Flutter code

So, what is extension in Dart?

Conclusion

Beginner Python tips Day - 07 Counter

Beginner Python tips Day - 06 try...except...else...finally

Beginner Python tips Day - 05 while...else, for...else

Beginner Python tips Day - 04 random

Beginner Python tips Day - 03 defaultdict

Beginner Python tips Day - 02 dict.get()

Beginner Python tips Day - 01 Enumerate

Microservice in Python using FastAPI

Creating Microservices with Python

Introduction to Microservices

Benefits of Microservice

Drawbacks of Microservice

Why Microservice in Python

Introduction to FastAPI

Why FastAPI

Installing FastAPI

Creating Simple REST API using FastAPI

Using PostgreSQL Database with FastAPI

Microservice Data Management Patterns

Database Per Service

Shared Database

API Composition

Creating a Python Microservice in Docker

Installing Docker and Docker Compose

Creating Movies Service

Creating Casts Service

Running the microservice using Docker Compose

Using Nginx to Access Both Services using a Single Host Address

Conclusion and Next Step

Flask Rest API -Part:6- Testing REST APIs

Part 6: Testing REST APIs

Why should we spend time writing tests?

Test Isolation

What we learned from this part of the series?

So, what is `extension` in Dart?